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. 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..cfd39ed91 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" @@ -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{ @@ -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