From 11a707983f0b2a6c43d785fa8a9a20a278b342e4 Mon Sep 17 00:00:00 2001 From: Miguel Martinez Trivino Date: Tue, 7 Jul 2026 11:54:23 +0200 Subject: [PATCH 1/3] fix(dispatcher): improve fanout delivery timeouts and retry budget - Bump MaxElapsedTime from 10s to 5m (~12-15 retry attempts) - Add per-attempt context.WithTimeout (10s) so a single hung HTTP request cannot consume the entire retry budget - Set http.Client.Timeout on the webhook plugin to match the per-attempt deadline (shared via sdk.PerAttemptTimeout) - Wrap backoff with backoff.WithContext so the retry loop stops promptly when the parent context is cancelled - Replace deprecated io/ioutil with io in the webhook plugin Refs #3269 Closes #39 Assisted-by: OpenCode Signed-off-by: Miguel Martinez Trivino Chainloop-Trace-Sessions: ses_0c4100537ffeM74j5m1U4O10fe --- .../dispatcher/dispatch_timeout_test.go | 183 ++++++++++++++++++ .../internal/dispatcher/dispatcher.go | 24 ++- .../plugins/core/webhook/v1/webhook.go | 8 +- app/controlplane/plugins/sdk/v1/fanout.go | 7 + 4 files changed, 214 insertions(+), 8 deletions(-) create mode 100644 app/controlplane/internal/dispatcher/dispatch_timeout_test.go diff --git a/app/controlplane/internal/dispatcher/dispatch_timeout_test.go b/app/controlplane/internal/dispatcher/dispatch_timeout_test.go new file mode 100644 index 000000000..66cebb343 --- /dev/null +++ b/app/controlplane/internal/dispatcher/dispatch_timeout_test.go @@ -0,0 +1,183 @@ +// +// Copyright 2026 The Chainloop Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package dispatcher + +import ( + "context" + "errors" + "io" + "testing" + "time" + + "github.com/chainloop-dev/chainloop/app/controlplane/plugins/sdk/v1" + mockedSDK "github.com/chainloop-dev/chainloop/app/controlplane/plugins/sdk/v1/mocks" + "github.com/go-kratos/kratos/v2/log" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// setMaxDispatchElapsedTimeForTest temporarily shrinks the total retry +// budget so tests don't wait for the real 5m. The returned func restores +// the production value. +func setMaxDispatchElapsedTimeForTest(d time.Duration) func() { + prev := maxDispatchElapsedTime + maxDispatchElapsedTime = d + return func() { maxDispatchElapsedTime = prev } +} + +// TestDispatchPerAttemptTimeout verifies that each Execute call runs under +// a per-attempt deadline rather than the unbounded context.TODO() that was +// previously passed through. The mock always fails so the retry loop fires +// multiple attempts; every observed context must carry a deadline within +// the perAttemptTimeout ballpark. +func TestDispatchPerAttemptTimeout(t *testing.T) { + // MaxElapsedTime must exceed the first backoff interval (~500ms ± jitter) + // so the retry loop fires at least 2 attempts. 2s yields 2-3 attempts + // with enough headroom to avoid flakes on loaded CI runners. + restore := setMaxDispatchElapsedTimeForTest(2 * time.Second) + t.Cleanup(restore) + + plugin := mockedSDK.NewFanOut(t) + plugin.On("String", mock.Anything).Return("test-plugin").Maybe() + + type attemptInfo struct { + hasDeadline bool + deadline time.Duration + } + var observed []attemptInfo + + plugin.On("Execute", mock.Anything, mock.Anything).Run(func(args mock.Arguments) { + ctx := args.Get(0).(context.Context) + deadline, ok := ctx.Deadline() + observed = append(observed, attemptInfo{ + hasDeadline: ok, + deadline: time.Until(deadline), + }) + }).Return(errors.New("transient failure")) + + logger := log.NewHelper(log.NewStdLogger(io.Discard)) + opts := &sdk.ExecutionRequest{ + Input: &sdk.ExecuteInput{ + Attestation: &sdk.ExecuteAttestation{}, + }, + } + + err := dispatch(context.Background(), plugin, opts, logger) + require.Error(t, err) + + require.GreaterOrEqual(t, len(observed), 2, "should have retried at least once") + + // Every attempt must have a per-attempt deadline near perAttemptTimeout. + // No attempt should inherit an unbounded (no-deadline) context. + for i, a := range observed { + assert.True(t, a.hasDeadline, "attempt %d had no deadline", i) + assert.Greater(t, a.deadline, 0*time.Second, "attempt %d deadline already exceeded", i) + assert.Less(t, a.deadline, sdk.PerAttemptTimeout+1*time.Second, + "attempt %d deadline %s exceeded perAttemptTimeout+slack", i, a.deadline) + } +} + +// TestDispatchSucceedsOnRetry verifies that a transient failure followed by +// success stops the retry loop and returns nil. +func TestDispatchSucceedsOnRetry(t *testing.T) { + // Safety net: if the mock setup is wrong, fail fast instead of waiting 5m. + restore := setMaxDispatchElapsedTimeForTest(2 * time.Second) + t.Cleanup(restore) + + plugin := mockedSDK.NewFanOut(t) + plugin.On("String", mock.Anything).Return("test-plugin").Maybe() + + // First call fails, second call succeeds. .Once() ensures each + // expectation is consumed exactly once, so the second call returns nil. + plugin.On("Execute", mock.Anything, mock.Anything).Return(errors.New("transient failure")).Once() + plugin.On("Execute", mock.Anything, mock.Anything).Return(nil).Once() + + logger := log.NewHelper(log.NewStdLogger(io.Discard)) + opts := &sdk.ExecutionRequest{ + Input: &sdk.ExecuteInput{ + Attestation: &sdk.ExecuteAttestation{}, + }, + } + + err := dispatch(context.Background(), plugin, opts, logger) + require.NoError(t, err) + + var execCalls int + for _, c := range plugin.Calls { + if c.Method == "Execute" { + execCalls++ + } + } + assert.Equal(t, 2, execCalls, "should have called Execute exactly twice") +} + +// TestDispatchNoInput verifies the fast-fail path that doesn't enter the retry loop. +func TestDispatchNoInput(t *testing.T) { + plugin := mockedSDK.NewFanOut(t) + logger := log.NewHelper(log.NewStdLogger(io.Discard)) + opts := &sdk.ExecutionRequest{Input: &sdk.ExecuteInput{}} + + err := dispatch(context.Background(), plugin, opts, logger) + require.Error(t, err) + assert.Contains(t, err.Error(), "no input provided") + plugin.AssertNotCalled(t, "Execute", mock.Anything, mock.Anything) +} + +// TestDispatchRespectsParentContextCancellation verifies that cancelling the +// parent context stops the retry loop promptly, rather than waiting for the +// full MaxElapsedTime budget to expire. +func TestDispatchRespectsParentContextCancellation(t *testing.T) { + restore := setMaxDispatchElapsedTimeForTest(30 * time.Second) + t.Cleanup(restore) + + plugin := mockedSDK.NewFanOut(t) + plugin.On("String", mock.Anything).Return("test-plugin").Maybe() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // Cancel the parent context on the first Execute call so the retry loop + // should stop during the backoff sleep that follows. + plugin.On("Execute", mock.Anything, mock.Anything).Run(func(_ mock.Arguments) { + cancel() + }).Return(errors.New("transient failure")) + + logger := log.NewHelper(log.NewStdLogger(io.Discard)) + opts := &sdk.ExecutionRequest{ + Input: &sdk.ExecuteInput{ + Attestation: &sdk.ExecuteAttestation{}, + }, + } + + done := make(chan error, 1) + go func() { + done <- dispatch(ctx, plugin, opts, logger) + }() + + select { + case err := <-done: + require.Error(t, err) + // The loop should exit with the parent context's cancellation error, + // not keep retrying for the full 30s budget. + assert.ErrorIs(t, err, context.Canceled) + case <-time.After(5 * time.Second): + cancel() // unblock the goroutine + err := <-done + assert.Failf(t, "dispatch did not return within 5s of parent context cancellation", + "eventually returned: %v", err) + } +} diff --git a/app/controlplane/internal/dispatcher/dispatcher.go b/app/controlplane/internal/dispatcher/dispatcher.go index 704bb6b2d..1d3408a5a 100644 --- a/app/controlplane/internal/dispatcher/dispatcher.go +++ b/app/controlplane/internal/dispatcher/dispatcher.go @@ -1,5 +1,5 @@ // -// Copyright 2023 The Chainloop Authors. +// Copyright 2023-2026 The Chainloop Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -48,6 +48,17 @@ type FanOutDispatcher struct { loaded sdk.AvailablePlugins } +// maxDispatchElapsedTime bounds the total time the dispatcher will keep +// retrying a single integration delivery. With the default exponential +// backoff (500ms initial, ×1.5 multiplier, 60s max interval) this yields +// roughly 12-15 attempts — enough to ride out a transient blip or a brief +// endpoint restart, while staying short enough not to leak goroutines in +// the current in-process, fire-and-forget dispatch model. +// +// It is a var (not a const) so tests can shrink it to avoid waiting for +// the real 5m budget. Production code must not mutate it. +var maxDispatchElapsedTime = 5 * time.Minute + func New(integrationUC *biz.IntegrationUseCase, wfUC *biz.WorkflowUseCase, wfRunUC *biz.WorkflowRunUseCase, creds credentials.ReaderWriter, c biz.CASClient, registered sdk.AvailablePlugins, l log.Logger) *FanOutDispatcher { return &FanOutDispatcher{integrationUC, wfUC, wfRunUC, creds, c, servicelogger.ScopedHelper(l, "fanout-dispatcher"), l, registered} } @@ -280,7 +291,7 @@ func (d *FanOutDispatcher) loadInputs(ctx context.Context, queue dispatchQueue, func dispatch(ctx context.Context, plugin sdk.FanOut, opts *sdk.ExecutionRequest, logger *log.Helper) error { b := backoff.NewExponentialBackOff() - b.MaxElapsedTime = 10 * time.Second + b.MaxElapsedTime = maxDispatchElapsedTime var inputType string switch { @@ -299,16 +310,21 @@ func dispatch(ctx context.Context, plugin sdk.FanOut, opts *sdk.ExecutionRequest return backoff.RetryNotify( func() error { + // Each attempt gets its own deadline so a single hung request + // cannot consume the entire retry budget. + attemptCtx, cancel := context.WithTimeout(ctx, sdk.PerAttemptTimeout) + defer cancel() + logger.Infow("msg", "executing integration", "integration", plugin.String(), "input", inputType) - err := plugin.Execute(ctx, opts) + err := plugin.Execute(attemptCtx, opts) if err == nil { logger.Infow("msg", "execution OK!", "integration", plugin.String(), "input", inputType) } return err }, - b, + backoff.WithContext(b, ctx), func(err error, delay time.Duration) { logger.Warnf("error executing integration %s, will retry in %s - %s", plugin.String(), delay, err) }, diff --git a/app/controlplane/plugins/core/webhook/v1/webhook.go b/app/controlplane/plugins/core/webhook/v1/webhook.go index afa83ac2e..e54add095 100644 --- a/app/controlplane/plugins/core/webhook/v1/webhook.go +++ b/app/controlplane/plugins/core/webhook/v1/webhook.go @@ -1,4 +1,4 @@ -// Copyright 2025 The Chainloop Authors. +// Copyright 2025-2026 The Chainloop Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -20,7 +20,7 @@ import ( "encoding/json" "errors" "fmt" - "io/ioutil" + "io" "net/http" "net/url" @@ -87,7 +87,7 @@ func New(l log.Logger) (sdk.FanOut, error) { return &Integration{ FanOutIntegration: base, - client: &http.Client{}, + client: &http.Client{Timeout: sdk.PerAttemptTimeout}, }, nil } @@ -259,7 +259,7 @@ func (i *Integration) sendWebhook(ctx context.Context, url, kind string, payload }() // Read response body for more detailed error messages - body, err := ioutil.ReadAll(resp.Body) + body, err := io.ReadAll(resp.Body) if err != nil { i.Logger.Warnw("failed to read response body", "error", err) } diff --git a/app/controlplane/plugins/sdk/v1/fanout.go b/app/controlplane/plugins/sdk/v1/fanout.go index d6e376309..0d5472896 100644 --- a/app/controlplane/plugins/sdk/v1/fanout.go +++ b/app/controlplane/plugins/sdk/v1/fanout.go @@ -33,6 +33,13 @@ import ( const IntegrationKindFanOut = "fan-out" +// PerAttemptTimeout is the hard per-attempt deadline the fanout dispatcher +// enforces on each Execute call. FanOut plugins that make HTTP calls should +// set their http.Client.Timeout to this value so a hung endpoint cannot +// consume the entire retry budget in a single attempt — the cutoff is +// enforced at whichever layer sees the stall first. +const PerAttemptTimeout = 10 * time.Second + // FanOut a FanOut is an integration for which we expect to fan out data to other systems type FanOut interface { // Integration has to be implemented by all integrations From e16b9b2d8d9d6dcd40368d0b94c731821cb21230 Mon Sep 17 00:00:00 2001 From: Miguel Martinez Trivino Date: Tue, 7 Jul 2026 11:56:58 +0200 Subject: [PATCH 2/3] fix(webhook): bump plugin version to 1.2 Reflects the new per-attempt HTTP timeout behavior introduced in the previous commit. Assisted-by: OpenCode Signed-off-by: Miguel Martinez Trivino Chainloop-Trace-Sessions: ses_0c4100537ffeM74j5m1U4O10fe --- app/controlplane/plugins/core/webhook/v1/webhook.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/app/controlplane/plugins/core/webhook/v1/webhook.go b/app/controlplane/plugins/core/webhook/v1/webhook.go index e54add095..cfd39ed91 100644 --- a/app/controlplane/plugins/core/webhook/v1/webhook.go +++ b/app/controlplane/plugins/core/webhook/v1/webhook.go @@ -69,7 +69,7 @@ func New(l log.Logger) (sdk.FanOut, error) { base, err := sdk.NewFanOut( &sdk.NewParams{ ID: "webhook", - Version: "1.1", + Version: "1.2", Description: "Send Attestation and SBOMs to a generic POST webhook URL", Logger: l, InputSchema: &sdk.InputSchema{ From 5bd7613c05ee57d3a0cf1613d9d02fe1f938783a Mon Sep 17 00:00:00 2001 From: Miguel Martinez Trivino Date: Tue, 7 Jul 2026 12:01:32 +0200 Subject: [PATCH 3/3] docs: clarify Assisted-by trailer must name the actual AI tool Add OpenCode to the examples and state explicitly that the value must match the tool that produced the work, not the underlying model, and must not be copied blindly from an example. Assisted-by: OpenCode Signed-off-by: Miguel Martinez Trivino Chainloop-Trace-Sessions: ses_0c4100537ffeM74j5m1U4O10fe --- AI_POLICY.md | 5 +++++ CLAUDE.md | 3 +++ 2 files changed, 8 insertions(+) diff --git a/AI_POLICY.md b/AI_POLICY.md index 395ee51fe..be42f8dd9 100644 --- a/AI_POLICY.md +++ b/AI_POLICY.md @@ -47,8 +47,13 @@ PR description. Add an `Assisted-by:` trailer to each affected commit: Assisted-by: GitHub Copilot Assisted-by: Claude Code Assisted-by: ChatGPT o3 +Assisted-by: OpenCode ``` +Use the name of the AI coding tool that actually produced the work (e.g. +`OpenCode`, `Claude Code`, `GitHub Copilot`), not the underlying model. Do not +blindly copy an example — pick the value that matches the tool you are running. + Disclosure is not a penalty — it is trust infrastructure. It preserves transparency, helps reviewers calibrate their attention, and keeps provenance clear for the project's long-term health. diff --git a/CLAUDE.md b/CLAUDE.md index 292b66887..77e0bec46 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -267,8 +267,11 @@ Code reviews are required for all submissions via GitHub pull requests. Assisted-by: GitHub Copilot Assisted-by: Claude Code Assisted-by: ChatGPT o3 + Assisted-by: OpenCode ``` + Use the name of the AI coding tool that actually produced the work (e.g. `OpenCode`, `Claude Code`, `GitHub Copilot`), not the underlying model. Do not blindly copy an example — pick the value that matches the tool you are running. + 2. **The PR description** MUST disclose the AI assistance. Apply this on the *first* commit/PR you create — do not wait for review feedback to add it. If you amend or rebase a commit that previously lacked the trailer, add it as part of the amend.