From eaacfdb253a9f043d49b0720f34b78e80ff33685 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Juli=C3=A1n=20Gonz=C3=A1lez?= Date: Thu, 6 Aug 2026 10:22:02 -0700 Subject: [PATCH 1/8] Poll legacy custom action managers to a terminal status registerLegacyAction discarded the id and status a CustomActionManager returned, so a non-terminal status resolved the outer action as complete with whatever partial response existed while the underlying action was still running. Keep the invoke response for synchronous managers that never populated id or status, and poll GetActionStatus with capped backoff for explicitly in-flight results, bounded by the handler context's deadline: tolerate a few consecutive lookup failures, carry the last real response, and fail the outer action when the inner one reports failure or an unexpected status. Co-Authored-By: Claude Fable 5 --- pkg/connectorbuilder/actions.go | 87 +++++++- pkg/connectorbuilder/actions_legacy_test.go | 227 ++++++++++++++++++++ 2 files changed, 311 insertions(+), 3 deletions(-) create mode 100644 pkg/connectorbuilder/actions_legacy_test.go diff --git a/pkg/connectorbuilder/actions.go b/pkg/connectorbuilder/actions.go index 314bdb782..5b8e93b55 100644 --- a/pkg/connectorbuilder/actions.go +++ b/pkg/connectorbuilder/actions.go @@ -5,12 +5,26 @@ import ( "fmt" "time" + "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap/ctxzap" + "go.uber.org/zap" + "google.golang.org/protobuf/types/known/structpb" + v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" "github.com/conductorone/baton-sdk/pkg/actions" "github.com/conductorone/baton-sdk/pkg/annotations" "github.com/conductorone/baton-sdk/pkg/types/tasks" "github.com/conductorone/baton-sdk/pkg/uotel" - "google.golang.org/protobuf/types/known/structpb" +) + +const ( + // maxConsecutiveStatusErrors bounds how many status-check failures in a + // row the legacy action poll loop tolerates before failing the action. + maxConsecutiveStatusErrors = 3 + + // The legacy status poll starts fast and backs off to a cap so a slow + // action doesn't drain a remote manager's rate-limit budget. + initialStatusPollInterval = time.Second + maxStatusPollInterval = 30 * time.Second ) // ActionManager defines the interface for managing actions in the connector builder. @@ -212,12 +226,79 @@ func (b *builder) GetActionStatus(ctx context.Context, request *v2.GetActionStat // registerLegacyAction wraps a legacy CustomActionManager action as an ActionHandler and registers it. func registerLegacyAction(ctx context.Context, registry actions.ActionRegistry, schema *v2.BatonActionSchema, legacyManager CustomActionManager) error { handler := func(ctx context.Context, args *structpb.Struct) (*structpb.Struct, annotations.Annotations, error) { - _, _, resp, annos, err := legacyManager.InvokeAction(ctx, schema.GetName(), "", args) - return resp, annos, err + // The inner call keeps the detached handler context; its one-hour + // deadline is the execution backstop for however long the legacy + // manager runs. + id, actionStatus, resp, annos, err := legacyManager.InvokeAction(ctx, schema.GetName(), "", args) + if err != nil { + return resp, annos, err + } + + // Only an explicitly in-flight status with a usable id is worth + // polling. Legacy managers were never required to populate id or + // status — the wrapper used to discard both — so anything else + // resolves the outer action with the response, as it always did. + if id == "" || !isInFlightActionStatus(actionStatus) { + return resp, annos, nil + } + + // Poll to a terminal status so the outer result carries the action's + // real outcome. A few consecutive status-check failures are tolerated: + // one flaky remote lookup must not convert a succeeding action into a + // failure. The interval backs off to a cap so a slow action doesn't + // drain a remote manager's rate-limit budget. + l := ctxzap.Extract(ctx) + statusErrs := 0 + interval := initialStatusPollInterval + timer := time.NewTimer(interval) + defer timer.Stop() + for { + select { + case <-ctx.Done(): + return resp, annos, ctx.Err() + case <-timer.C: + } + interval = min(interval*2, maxStatusPollInterval) + timer.Reset(interval) + + st, _, pollResp, pollAnnos, err := legacyManager.GetActionStatus(ctx, id) + if err != nil { + statusErrs++ + if statusErrs >= maxConsecutiveStatusErrors { + return resp, annos, err + } + l.Warn("legacy action status check failed, retrying", + zap.String("action", schema.GetName()), + zap.Int("consecutive_errors", statusErrs), + zap.Error(err)) + continue + } + statusErrs = 0 + + // An in-flight poll may carry no response; keep the last real one + // so the error exits above still return it. + if pollResp != nil { + resp, annos = pollResp, pollAnnos + } + + switch { + case isInFlightActionStatus(st): + case st == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE: + return resp, annos, nil + case st == v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED: + return resp, annos, fmt.Errorf("legacy action %q failed", schema.GetName()) + default: + return resp, annos, fmt.Errorf("legacy action %q returned unexpected status %s", schema.GetName(), st.String()) + } + } } return registry.Register(ctx, schema, handler) } +func isInFlightActionStatus(s v2.BatonActionStatus) bool { + return s == v2.BatonActionStatus_BATON_ACTION_STATUS_PENDING || s == v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING +} + // addActionManager handles deprecated CustomActionManager and RegisterActionManagerLimited interfaces // by extracting their actions and registering them into the unified ActionManager. func (b *builder) addActionManager(ctx context.Context, in interface{}, registry actions.ActionRegistry) error { diff --git a/pkg/connectorbuilder/actions_legacy_test.go b/pkg/connectorbuilder/actions_legacy_test.go new file mode 100644 index 000000000..27fd4c492 --- /dev/null +++ b/pkg/connectorbuilder/actions_legacy_test.go @@ -0,0 +1,227 @@ +package connectorbuilder + +import ( + "context" + "sync/atomic" + "testing" + "time" + + v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" + "github.com/conductorone/baton-sdk/pkg/actions" + "github.com/conductorone/baton-sdk/pkg/annotations" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/types/known/structpb" +) + +type fakeThirdPartyActionManager struct { + schema *v2.BatonActionSchema + invokeDeadline time.Time + hadDeadline bool + invoked chan struct{} +} + +func (f *fakeThirdPartyActionManager) ListActionSchemas(_ context.Context, _ string) ([]*v2.BatonActionSchema, annotations.Annotations, error) { + return []*v2.BatonActionSchema{f.schema}, nil, nil +} + +func (f *fakeThirdPartyActionManager) GetActionSchema(_ context.Context, _ string) (*v2.BatonActionSchema, annotations.Annotations, error) { + return f.schema, nil, nil +} + +func (f *fakeThirdPartyActionManager) InvokeAction(ctx context.Context, _ string, _ string, _ *structpb.Struct) (string, v2.BatonActionStatus, *structpb.Struct, annotations.Annotations, error) { + f.invokeDeadline, f.hadDeadline = ctx.Deadline() + close(f.invoked) + return "legacy-1", v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE, nil, nil, nil +} + +func (f *fakeThirdPartyActionManager) GetActionStatus(_ context.Context, _ string) (v2.BatonActionStatus, string, *structpb.Struct, annotations.Annotations, error) { + return v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE, "legacy", nil, nil, nil +} + +// A third-party CustomActionManager must receive the detached handler context +// (one-hour deadline), not the 2s inline-wait pin reserved for the SDK's own +// deadline-aware ActionManager: it treats the deadline as an execution cap. +func TestRegisterLegacyActionThirdPartyManagerKeepsHandlerContext(t *testing.T) { + ctx := t.Context() + + legacy := &fakeThirdPartyActionManager{ + schema: v2.BatonActionSchema_builder{Name: "legacy_action"}.Build(), + invoked: make(chan struct{}), + } + + m := actions.NewActionManager(ctx) + require.NoError(t, registerLegacyAction(ctx, m, legacy.schema, legacy)) + + _, _, _, _, err := m.InvokeAction(ctx, "legacy_action", "", &structpb.Struct{}) + require.NoError(t, err) + + select { + case <-legacy.invoked: + case <-time.After(5 * time.Second): + t.Fatal("legacy manager was never invoked") + } + + require.True(t, legacy.hadDeadline) + require.Greater(t, time.Until(legacy.invokeDeadline), 30*time.Minute) +} + +// A deadline-aware inner ActionManager blocks until the action truly +// finishes: the outer action must stay RUNNING at its own inline wait and +// resolve later with the real response, never as an empty completion. +func TestRegisterLegacyActionTracksInnerManagerToCompletion(t *testing.T) { + ctx := t.Context() + + schema := v2.BatonActionSchema_builder{Name: "inner_action"}.Build() + rv, err := structpb.NewStruct(map[string]any{"done": true}) + require.NoError(t, err) + + inner := actions.NewActionManager(ctx) + require.NoError(t, inner.Register(ctx, schema, func(_ context.Context, _ *structpb.Struct) (*structpb.Struct, annotations.Annotations, error) { + time.Sleep(1500 * time.Millisecond) + return rv, nil, nil + })) + + outer := actions.NewActionManager(ctx) + require.NoError(t, registerLegacyAction(ctx, outer, schema, inner)) + + outerID, outerStatus, outerRv, _, err := outer.InvokeAction(ctx, "inner_action", "", &structpb.Struct{}) + require.NoError(t, err) + require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING, outerStatus) + require.Nil(t, outerRv) + + require.Eventually(t, func() bool { + st, _, gotRv, _, err := outer.GetActionStatus(ctx, outerID) + return err == nil && st == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE && gotRv != nil + }, 5*time.Second, 100*time.Millisecond) +} + +type fakeAsyncThirdPartyActionManager struct { + schema *v2.BatonActionSchema + rv *structpb.Struct + finalStatus v2.BatonActionStatus + statusCalls atomic.Int32 + gotID atomic.Value +} + +func (f *fakeAsyncThirdPartyActionManager) ListActionSchemas(_ context.Context, _ string) ([]*v2.BatonActionSchema, annotations.Annotations, error) { + return []*v2.BatonActionSchema{f.schema}, nil, nil +} + +func (f *fakeAsyncThirdPartyActionManager) GetActionSchema(_ context.Context, _ string) (*v2.BatonActionSchema, annotations.Annotations, error) { + return f.schema, nil, nil +} + +func (f *fakeAsyncThirdPartyActionManager) InvokeAction(_ context.Context, _ string, _ string, _ *structpb.Struct) (string, v2.BatonActionStatus, *structpb.Struct, annotations.Annotations, error) { + return "legacy-async-1", v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING, nil, nil, nil +} + +func (f *fakeAsyncThirdPartyActionManager) GetActionStatus(_ context.Context, id string) (v2.BatonActionStatus, string, *structpb.Struct, annotations.Annotations, error) { + f.gotID.Store(id) + switch f.statusCalls.Add(1) { + case 1: + // One transient lookup failure must not fail the action. + return v2.BatonActionStatus_BATON_ACTION_STATUS_UNKNOWN, "", nil, nil, context.DeadlineExceeded + case 2: + return v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING, "async", nil, nil, nil + default: + return f.finalStatus, "async", f.rv, nil, nil + } +} + +// A third-party manager returning a non-terminal status gets polled to a +// terminal one with the action id it returned, riding through a transient +// status-check failure. +func TestRegisterLegacyActionPollsAsyncThirdPartyManager(t *testing.T) { + ctx := t.Context() + + rv, err := structpb.NewStruct(map[string]any{"done": true}) + require.NoError(t, err) + legacy := &fakeAsyncThirdPartyActionManager{ + schema: v2.BatonActionSchema_builder{Name: "async_action"}.Build(), + rv: rv, + finalStatus: v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE, + } + + outer := actions.NewActionManager(ctx) + require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy)) + + outerID, _, _, _, err := outer.InvokeAction(ctx, "async_action", "", &structpb.Struct{}) + require.NoError(t, err) + + // Backoff polls land at roughly 1s, 3s, and 7s. + require.Eventually(t, func() bool { + st, _, gotRv, _, err := outer.GetActionStatus(ctx, outerID) + return err == nil && st == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE && gotRv != nil + }, 20*time.Second, 100*time.Millisecond) + + require.Equal(t, "legacy-async-1", legacy.gotID.Load()) + require.GreaterOrEqual(t, legacy.statusCalls.Load(), int32(3)) +} + +// A legacy action that polls to FAILED must mark the outer action FAILED, +// never resolve it as a success. +func TestRegisterLegacyActionPolledFailureFailsOuterAction(t *testing.T) { + ctx := t.Context() + + legacy := &fakeAsyncThirdPartyActionManager{ + schema: v2.BatonActionSchema_builder{Name: "failing_action"}.Build(), + finalStatus: v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED, + } + + outer := actions.NewActionManager(ctx) + require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy)) + + outerID, _, _, _, err := outer.InvokeAction(ctx, "failing_action", "", &structpb.Struct{}) + require.NoError(t, err) + + require.Eventually(t, func() bool { + st, _, _, _, err := outer.GetActionStatus(ctx, outerID) + return err == nil && st == v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED + }, 20*time.Second, 100*time.Millisecond) +} + +type fakeSyncNoStatusActionManager struct { + schema *v2.BatonActionSchema + rv *structpb.Struct + statusCalls atomic.Int32 +} + +func (f *fakeSyncNoStatusActionManager) ListActionSchemas(_ context.Context, _ string) ([]*v2.BatonActionSchema, annotations.Annotations, error) { + return []*v2.BatonActionSchema{f.schema}, nil, nil +} + +func (f *fakeSyncNoStatusActionManager) GetActionSchema(_ context.Context, _ string) (*v2.BatonActionSchema, annotations.Annotations, error) { + return f.schema, nil, nil +} + +func (f *fakeSyncNoStatusActionManager) InvokeAction(_ context.Context, _ string, _ string, _ *structpb.Struct) (string, v2.BatonActionStatus, *structpb.Struct, annotations.Annotations, error) { + return "", v2.BatonActionStatus_BATON_ACTION_STATUS_UNSPECIFIED, f.rv, nil, nil +} + +func (f *fakeSyncNoStatusActionManager) GetActionStatus(_ context.Context, _ string) (v2.BatonActionStatus, string, *structpb.Struct, annotations.Annotations, error) { + f.statusCalls.Add(1) + return v2.BatonActionStatus_BATON_ACTION_STATUS_UNSPECIFIED, "", nil, nil, nil +} + +// Legacy synchronous managers were never required to populate id or status; +// a response with the zero status and no id must resolve the outer action +// immediately, never enter the polling loop. +func TestRegisterLegacyActionSyncManagerWithoutStatusResolves(t *testing.T) { + ctx := t.Context() + + rv, err := structpb.NewStruct(map[string]any{"done": true}) + require.NoError(t, err) + legacy := &fakeSyncNoStatusActionManager{ + schema: v2.BatonActionSchema_builder{Name: "sync_action"}.Build(), + rv: rv, + } + + outer := actions.NewActionManager(ctx) + require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy)) + + _, outerStatus, outerRv, _, err := outer.InvokeAction(ctx, "sync_action", "", &structpb.Struct{}) + require.NoError(t, err) + require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE, outerStatus) + require.NotNil(t, outerRv) + require.Equal(t, int32(0), legacy.statusCalls.Load()) +} From a786b9322419135b5c91a8870741d950073fc60c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Juli=C3=A1n=20Gonz=C3=A1lez?= Date: Mon, 10 Aug 2026 16:12:10 -0700 Subject: [PATCH 2/8] Resolve terminal invoke statuses at the seam and harden the poll loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A legacy manager reporting failure in-band — FAILED status with a nil error, as the SDK's own manager does for a fast-failing handler — resolved the outer action as a success at the invoke seam. Terminal statuses at the seam now resolve exactly like terminal polls, with the status checked before the id so an explicit failure without an id cannot pass through either; only a never-populated status keeps the legacy fire-and-forget behavior. The poll loop treats indeterminate statuses with the same three-strike tolerance as lookup errors, keeps only meaningful poll payloads, and reports the handler deadline's cause instead of a bare context error. The global invoke path now carries the caller's logger into the detached handler context, matching the resource path, so the loop's warnings are no longer dropped; an observer-backed test pins that. Poll intervals are variables so tests drive the loop in milliseconds: an outcome table covers the seam and poll matrices, and the original interval-driven tests shed about fifteen seconds of suite wall time. Co-Authored-By: Claude Fable 5 --- pkg/actions/actions.go | 1 + pkg/connectorbuilder/actions.go | 76 +++++-- pkg/connectorbuilder/actions_legacy_test.go | 159 +++++++++++++- .../zap/zaptest/observer/logged_entry.go | 39 ++++ .../zap/zaptest/observer/observer.go | 203 ++++++++++++++++++ vendor/modules.txt | 1 + 6 files changed, 456 insertions(+), 23 deletions(-) create mode 100644 vendor/go.uber.org/zap/zaptest/observer/logged_entry.go create mode 100644 vendor/go.uber.org/zap/zaptest/observer/observer.go diff --git a/pkg/actions/actions.go b/pkg/actions/actions.go index c7ef749df..9b35bb55a 100644 --- a/pkg/actions/actions.go +++ b/pkg/actions/actions.go @@ -636,6 +636,7 @@ func (a *ActionManager) invokeGlobalAction( }() oa.SetStatus(ctx, v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING) bgCtx := trace.ContextWithSpanContext(context.Background(), trace.SpanContextFromContext(ctx)) + bgCtx = ctxzap.ToContext(bgCtx, ctxzap.Extract(ctx)) handlerCtx, cancel := context.WithTimeoutCause(bgCtx, 1*time.Hour, errors.New("action handler timed out")) defer cancel() rv, annos, oaErr := handler(handlerCtx, args) diff --git a/pkg/connectorbuilder/actions.go b/pkg/connectorbuilder/actions.go index 5b8e93b55..aed89acb1 100644 --- a/pkg/connectorbuilder/actions.go +++ b/pkg/connectorbuilder/actions.go @@ -17,12 +17,16 @@ import ( ) const ( - // maxConsecutiveStatusErrors bounds how many status-check failures in a - // row the legacy action poll loop tolerates before failing the action. + // maxConsecutiveStatusErrors bounds how many status-check failures or + // indeterminate statuses in a row the legacy action poll loop tolerates + // before failing the action. maxConsecutiveStatusErrors = 3 +) - // The legacy status poll starts fast and backs off to a cap so a slow - // action doesn't drain a remote manager's rate-limit budget. +// The legacy status poll starts fast and backs off to a cap so a slow +// action doesn't drain a remote manager's rate-limit budget. Variables so +// tests can drive the loop without real-time waits. +var ( initialStatusPollInterval = time.Second maxStatusPollInterval = 30 * time.Second ) @@ -234,11 +238,23 @@ func registerLegacyAction(ctx context.Context, registry actions.ActionRegistry, return resp, annos, err } - // Only an explicitly in-flight status with a usable id is worth - // polling. Legacy managers were never required to populate id or - // status — the wrapper used to discard both — so anything else + // Legacy managers were never required to populate id or status — the + // wrapper used to discard both — so the never-populated shape // resolves the outer action with the response, as it always did. - if id == "" || !isInFlightActionStatus(actionStatus) { + if actionStatus == v2.BatonActionStatus_BATON_ACTION_STATUS_UNSPECIFIED { + return resp, annos, nil + } + + // A terminal status at the invoke seam resolves like a terminal + // poll. The SDK's own manager reports handler failures in-band as + // FAILED with a nil error, so this must not resolve as success. + if !isInFlightActionStatus(actionStatus) { + return resp, annos, legacyStatusErr(schema.GetName(), actionStatus) + } + + // An in-flight claim without an id cannot be polled; resolve with + // the response, matching the old fire-and-forget behavior. + if id == "" { return resp, annos, nil } @@ -255,7 +271,7 @@ func registerLegacyAction(ctx context.Context, registry actions.ActionRegistry, for { select { case <-ctx.Done(): - return resp, annos, ctx.Err() + return resp, annos, fmt.Errorf("legacy action %q did not reach a terminal status: %w", schema.GetName(), context.Cause(ctx)) case <-timer.C: } interval = min(interval*2, maxStatusPollInterval) @@ -269,32 +285,52 @@ func registerLegacyAction(ctx context.Context, registry actions.ActionRegistry, } l.Warn("legacy action status check failed, retrying", zap.String("action", schema.GetName()), - zap.Int("consecutive_errors", statusErrs), + zap.Int("consecutive_anomalies", statusErrs), zap.Error(err)) continue } - statusErrs = 0 - - // An in-flight poll may carry no response; keep the last real one - // so the error exits above still return it. - if pollResp != nil { + // Keep the last meaningful response for the error exits above; + // an indeterminate poll's payload must not replace it. + if pollResp != nil && (isInFlightActionStatus(st) || st == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE || st == v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED) { resp, annos = pollResp, pollAnnos } switch { case isInFlightActionStatus(st): - case st == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE: - return resp, annos, nil - case st == v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED: - return resp, annos, fmt.Errorf("legacy action %q failed", schema.GetName()) + statusErrs = 0 + case st == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE || st == v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED: + return resp, annos, legacyStatusErr(schema.GetName(), st) default: - return resp, annos, fmt.Errorf("legacy action %q returned unexpected status %s", schema.GetName(), st.String()) + // An indeterminate status gets the same tolerance as a + // lookup error: transient anomalies recover, persistent + // ones fail closed. + statusErrs++ + if statusErrs >= maxConsecutiveStatusErrors { + return resp, annos, fmt.Errorf("legacy action %q returned unexpected status %s", schema.GetName(), st.String()) + } + l.Warn("legacy action returned indeterminate status, retrying", + zap.String("action", schema.GetName()), + zap.String("status", st.String()), + zap.Int("consecutive_anomalies", statusErrs)) } } } return registry.Register(ctx, schema, handler) } +// legacyStatusErr maps a settled legacy status to the outer handler error: +// COMPLETE resolves clean, anything else fails the action. +func legacyStatusErr(name string, st v2.BatonActionStatus) error { + switch st { + case v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE: + return nil + case v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED: + return fmt.Errorf("legacy action %q failed", name) + default: + return fmt.Errorf("legacy action %q returned unexpected status %s", name, st.String()) + } +} + func isInFlightActionStatus(s v2.BatonActionStatus) bool { return s == v2.BatonActionStatus_BATON_ACTION_STATUS_PENDING || s == v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING } diff --git a/pkg/connectorbuilder/actions_legacy_test.go b/pkg/connectorbuilder/actions_legacy_test.go index 27fd4c492..bed6ccf58 100644 --- a/pkg/connectorbuilder/actions_legacy_test.go +++ b/pkg/connectorbuilder/actions_legacy_test.go @@ -2,6 +2,7 @@ package connectorbuilder import ( "context" + "fmt" "sync/atomic" "testing" "time" @@ -9,7 +10,10 @@ import ( v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" "github.com/conductorone/baton-sdk/pkg/actions" "github.com/conductorone/baton-sdk/pkg/annotations" + "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap/ctxzap" "github.com/stretchr/testify/require" + "go.uber.org/zap" + "go.uber.org/zap/zaptest/observer" "google.golang.org/protobuf/types/known/structpb" ) @@ -69,6 +73,7 @@ func TestRegisterLegacyActionThirdPartyManagerKeepsHandlerContext(t *testing.T) // finishes: the outer action must stay RUNNING at its own inline wait and // resolve later with the real response, never as an empty completion. func TestRegisterLegacyActionTracksInnerManagerToCompletion(t *testing.T) { + shortStatusPolls(t) ctx := t.Context() schema := v2.BatonActionSchema_builder{Name: "inner_action"}.Build() @@ -132,6 +137,7 @@ func (f *fakeAsyncThirdPartyActionManager) GetActionStatus(_ context.Context, id // terminal one with the action id it returned, riding through a transient // status-check failure. func TestRegisterLegacyActionPollsAsyncThirdPartyManager(t *testing.T) { + shortStatusPolls(t) ctx := t.Context() rv, err := structpb.NewStruct(map[string]any{"done": true}) @@ -148,11 +154,10 @@ func TestRegisterLegacyActionPollsAsyncThirdPartyManager(t *testing.T) { outerID, _, _, _, err := outer.InvokeAction(ctx, "async_action", "", &structpb.Struct{}) require.NoError(t, err) - // Backoff polls land at roughly 1s, 3s, and 7s. require.Eventually(t, func() bool { st, _, gotRv, _, err := outer.GetActionStatus(ctx, outerID) return err == nil && st == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE && gotRv != nil - }, 20*time.Second, 100*time.Millisecond) + }, 5*time.Second, 10*time.Millisecond) require.Equal(t, "legacy-async-1", legacy.gotID.Load()) require.GreaterOrEqual(t, legacy.statusCalls.Load(), int32(3)) @@ -161,6 +166,7 @@ func TestRegisterLegacyActionPollsAsyncThirdPartyManager(t *testing.T) { // A legacy action that polls to FAILED must mark the outer action FAILED, // never resolve it as a success. func TestRegisterLegacyActionPolledFailureFailsOuterAction(t *testing.T) { + shortStatusPolls(t) ctx := t.Context() legacy := &fakeAsyncThirdPartyActionManager{ @@ -177,7 +183,7 @@ func TestRegisterLegacyActionPolledFailureFailsOuterAction(t *testing.T) { require.Eventually(t, func() bool { st, _, _, _, err := outer.GetActionStatus(ctx, outerID) return err == nil && st == v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED - }, 20*time.Second, 100*time.Millisecond) + }, 5*time.Second, 10*time.Millisecond) } type fakeSyncNoStatusActionManager struct { @@ -225,3 +231,150 @@ func TestRegisterLegacyActionSyncManagerWithoutStatusResolves(t *testing.T) { require.NotNil(t, outerRv) require.Equal(t, int32(0), legacy.statusCalls.Load()) } + +// shortStatusPolls drives the poll loop in milliseconds so outcome tables +// don't add real wall time to the suite. +func shortStatusPolls(t *testing.T) { + t.Helper() + origInitial, origMax := initialStatusPollInterval, maxStatusPollInterval + initialStatusPollInterval, maxStatusPollInterval = time.Millisecond, 4*time.Millisecond + t.Cleanup(func() { + initialStatusPollInterval, maxStatusPollInterval = origInitial, origMax + }) +} + +type scriptedPollResult struct { + status v2.BatonActionStatus + err error +} + +type scriptedLegacyActionManager struct { + schema *v2.BatonActionSchema + invokeID string + invokeStatus v2.BatonActionStatus + invokeRv *structpb.Struct + polls []scriptedPollResult + pollCalls atomic.Int32 +} + +func (f *scriptedLegacyActionManager) ListActionSchemas(_ context.Context, _ string) ([]*v2.BatonActionSchema, annotations.Annotations, error) { + return []*v2.BatonActionSchema{f.schema}, nil, nil +} + +func (f *scriptedLegacyActionManager) GetActionSchema(_ context.Context, _ string) (*v2.BatonActionSchema, annotations.Annotations, error) { + return f.schema, nil, nil +} + +func (f *scriptedLegacyActionManager) InvokeAction(_ context.Context, _ string, _ string, _ *structpb.Struct) (string, v2.BatonActionStatus, *structpb.Struct, annotations.Annotations, error) { + return f.invokeID, f.invokeStatus, f.invokeRv, nil, nil +} + +func (f *scriptedLegacyActionManager) GetActionStatus(_ context.Context, _ string) (v2.BatonActionStatus, string, *structpb.Struct, annotations.Annotations, error) { + i := int(f.pollCalls.Add(1)) - 1 + if i >= len(f.polls) { + // Polling past the script (or a case that should never poll) fails + // loudly instead of hanging the loop. + return v2.BatonActionStatus_BATON_ACTION_STATUS_UNSPECIFIED, "", nil, nil, fmt.Errorf("unexpected status poll %d", i+1) + } + p := f.polls[i] + if p.err != nil { + return v2.BatonActionStatus_BATON_ACTION_STATUS_UNKNOWN, "", nil, nil, p.err + } + return p.status, "scripted", f.invokeRv, nil, nil +} + +// Every seam and poll outcome the wrapper distinguishes: terminal statuses at +// the invoke seam resolve like terminal polls (in-band FAILED must not become +// a success), never-populated shapes keep the legacy pass-through, and the +// poll loop tolerates transient lookup errors and indeterminate statuses up +// to the shared threshold before failing closed. +func TestRegisterLegacyActionSeamAndPollOutcomes(t *testing.T) { + shortStatusPolls(t) + + const ( + unspecified = v2.BatonActionStatus_BATON_ACTION_STATUS_UNSPECIFIED + unknown = v2.BatonActionStatus_BATON_ACTION_STATUS_UNKNOWN + running = v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING + complete = v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE + failed = v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED + ) + lookupErr := scriptedPollResult{err: context.DeadlineExceeded} + + cases := []struct { + name string + invokeID string + invokeStatus v2.BatonActionStatus + polls []scriptedPollResult + wantStatus v2.BatonActionStatus + wantErrIn string + }{ + {"in-band failure at the invoke seam fails", "id-1", failed, nil, failed, "failed"}, + {"unexpected status at the invoke seam fails", "id-1", unknown, nil, failed, "unexpected status"}, + {"unspecified status resolves as before", "id-1", unspecified, nil, complete, ""}, + {"in-flight without an id resolves as before", "", running, nil, complete, ""}, + {"threshold consecutive lookup errors fail closed", "id-1", running, + []scriptedPollResult{lookupErr, lookupErr, lookupErr}, failed, "deadline"}, + {"lookup errors under the threshold recover", "id-1", running, + []scriptedPollResult{lookupErr, lookupErr, {status: complete}}, complete, ""}, + {"indeterminate polls under the threshold recover", "id-1", running, + []scriptedPollResult{{status: unspecified}, {status: running}, {status: complete}}, complete, ""}, + {"persistent indeterminate polls fail closed", "id-1", running, + []scriptedPollResult{{status: unspecified}, {status: unspecified}, {status: unspecified}}, failed, "unexpected status"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + ctx := t.Context() + rv, err := structpb.NewStruct(map[string]any{"done": true}) + require.NoError(t, err) + legacy := &scriptedLegacyActionManager{ + schema: v2.BatonActionSchema_builder{Name: "scripted_action"}.Build(), + invokeID: tc.invokeID, + invokeStatus: tc.invokeStatus, + invokeRv: rv, + polls: tc.polls, + } + outer := actions.NewActionManager(ctx) + require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy)) + + _, st, gotRv, _, err := outer.InvokeAction(ctx, "scripted_action", "", &structpb.Struct{}) + require.NoError(t, err) + require.Equal(t, tc.wantStatus, st) + if tc.wantErrIn != "" { + require.Contains(t, gotRv.Fields["error"].GetStringValue(), tc.wantErrIn) + } else { + require.Nil(t, gotRv.Fields["error"]) + } + }) + } +} + +// The poll loop's warnings must reach the caller's logger: the detached +// handler context carries it across the goroutine boundary. +func TestLegacyPollWarningsReachCallerLogger(t *testing.T) { + shortStatusPolls(t) + + core, observed := observer.New(zap.WarnLevel) + ctx := ctxzap.ToContext(t.Context(), zap.New(core)) + + rv, err := structpb.NewStruct(map[string]any{"done": true}) + require.NoError(t, err) + legacy := &scriptedLegacyActionManager{ + schema: v2.BatonActionSchema_builder{Name: "warned_action"}.Build(), + invokeID: "id-1", + invokeStatus: v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING, + invokeRv: rv, + polls: []scriptedPollResult{ + {err: context.DeadlineExceeded}, + {status: v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE}, + }, + } + outer := actions.NewActionManager(ctx) + require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy)) + + _, st, _, _, err := outer.InvokeAction(ctx, "warned_action", "", &structpb.Struct{}) + require.NoError(t, err) + require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE, st) + require.Eventually(t, func() bool { + return observed.FilterMessage("legacy action status check failed, retrying").Len() == 1 + }, time.Second, 10*time.Millisecond) +} diff --git a/vendor/go.uber.org/zap/zaptest/observer/logged_entry.go b/vendor/go.uber.org/zap/zaptest/observer/logged_entry.go new file mode 100644 index 000000000..ef89e25c3 --- /dev/null +++ b/vendor/go.uber.org/zap/zaptest/observer/logged_entry.go @@ -0,0 +1,39 @@ +// Copyright (c) 2017 Uber Technologies, Inc. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +package observer + +import "go.uber.org/zap/zapcore" + +// A LoggedEntry is an encoding-agnostic representation of a log message. +// Field availability is context dependent. +type LoggedEntry struct { + zapcore.Entry + Context []zapcore.Field +} + +// ContextMap returns a map for all fields in Context. +func (e LoggedEntry) ContextMap() map[string]interface{} { + encoder := zapcore.NewMapObjectEncoder() + for _, f := range e.Context { + f.AddTo(encoder) + } + return encoder.Fields +} diff --git a/vendor/go.uber.org/zap/zaptest/observer/observer.go b/vendor/go.uber.org/zap/zaptest/observer/observer.go new file mode 100644 index 000000000..4f7ce0ec6 --- /dev/null +++ b/vendor/go.uber.org/zap/zaptest/observer/observer.go @@ -0,0 +1,203 @@ +// Copyright (c) 2016-2022 Uber Technologies, Inc. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +// Package observer provides a zapcore.Core that keeps an in-memory, +// encoding-agnostic representation of log entries. It's useful for +// applications that want to unit test their log output without tying their +// tests to a particular output encoding. +package observer // import "go.uber.org/zap/zaptest/observer" + +import ( + "strings" + "sync" + "time" + + "go.uber.org/zap/internal" + "go.uber.org/zap/zapcore" +) + +// ObservedLogs is a concurrency-safe, ordered collection of observed logs. +type ObservedLogs struct { + mu sync.RWMutex + logs []LoggedEntry +} + +// Len returns the number of items in the collection. +func (o *ObservedLogs) Len() int { + o.mu.RLock() + n := len(o.logs) + o.mu.RUnlock() + return n +} + +// All returns a copy of all the observed logs. +func (o *ObservedLogs) All() []LoggedEntry { + o.mu.RLock() + ret := make([]LoggedEntry, len(o.logs)) + copy(ret, o.logs) + o.mu.RUnlock() + return ret +} + +// TakeAll returns a copy of all the observed logs, and truncates the observed +// slice. +func (o *ObservedLogs) TakeAll() []LoggedEntry { + o.mu.Lock() + ret := o.logs + o.logs = nil + o.mu.Unlock() + return ret +} + +// AllUntimed returns a copy of all the observed logs, but overwrites the +// observed timestamps with time.Time's zero value. This is useful when making +// assertions in tests. +func (o *ObservedLogs) AllUntimed() []LoggedEntry { + ret := o.All() + for i := range ret { + ret[i].Time = time.Time{} + } + return ret +} + +// FilterLevelExact filters entries to those logged at exactly the given level. +func (o *ObservedLogs) FilterLevelExact(level zapcore.Level) *ObservedLogs { + return o.Filter(func(e LoggedEntry) bool { + return e.Level == level + }) +} + +// FilterMessage filters entries to those that have the specified message. +func (o *ObservedLogs) FilterMessage(msg string) *ObservedLogs { + return o.Filter(func(e LoggedEntry) bool { + return e.Message == msg + }) +} + +// FilterLoggerName filters entries to those logged through logger with the specified logger name. +func (o *ObservedLogs) FilterLoggerName(name string) *ObservedLogs { + return o.Filter(func(e LoggedEntry) bool { + return e.LoggerName == name + }) +} + +// FilterMessageSnippet filters entries to those that have a message containing the specified snippet. +func (o *ObservedLogs) FilterMessageSnippet(snippet string) *ObservedLogs { + return o.Filter(func(e LoggedEntry) bool { + return strings.Contains(e.Message, snippet) + }) +} + +// FilterField filters entries to those that have the specified field. +func (o *ObservedLogs) FilterField(field zapcore.Field) *ObservedLogs { + return o.Filter(func(e LoggedEntry) bool { + for _, ctxField := range e.Context { + if ctxField.Equals(field) { + return true + } + } + return false + }) +} + +// FilterFieldKey filters entries to those that have the specified key. +func (o *ObservedLogs) FilterFieldKey(key string) *ObservedLogs { + return o.Filter(func(e LoggedEntry) bool { + for _, ctxField := range e.Context { + if ctxField.Key == key { + return true + } + } + return false + }) +} + +// Filter returns a copy of this ObservedLogs containing only those entries +// for which the provided function returns true. +func (o *ObservedLogs) Filter(keep func(LoggedEntry) bool) *ObservedLogs { + o.mu.RLock() + defer o.mu.RUnlock() + + var filtered []LoggedEntry + for _, entry := range o.logs { + if keep(entry) { + filtered = append(filtered, entry) + } + } + return &ObservedLogs{logs: filtered} +} + +func (o *ObservedLogs) add(log LoggedEntry) { + o.mu.Lock() + o.logs = append(o.logs, log) + o.mu.Unlock() +} + +// New creates a new Core that buffers logs in memory (without any encoding). +// It's particularly useful in tests. +func New(enab zapcore.LevelEnabler) (zapcore.Core, *ObservedLogs) { + ol := &ObservedLogs{} + return &contextObserver{ + LevelEnabler: enab, + logs: ol, + }, ol +} + +type contextObserver struct { + zapcore.LevelEnabler + logs *ObservedLogs + context []zapcore.Field +} + +var ( + _ zapcore.Core = (*contextObserver)(nil) + _ internal.LeveledEnabler = (*contextObserver)(nil) +) + +func (co *contextObserver) Level() zapcore.Level { + return zapcore.LevelOf(co.LevelEnabler) +} + +func (co *contextObserver) Check(ent zapcore.Entry, ce *zapcore.CheckedEntry) *zapcore.CheckedEntry { + if co.Enabled(ent.Level) { + return ce.AddCore(ent, co) + } + return ce +} + +func (co *contextObserver) With(fields []zapcore.Field) zapcore.Core { + return &contextObserver{ + LevelEnabler: co.LevelEnabler, + logs: co.logs, + context: append(co.context[:len(co.context):len(co.context)], fields...), + } +} + +func (co *contextObserver) Write(ent zapcore.Entry, fields []zapcore.Field) error { + all := make([]zapcore.Field, 0, len(fields)+len(co.context)) + all = append(all, co.context...) + all = append(all, fields...) + co.logs.add(LoggedEntry{ent, all}) + return nil +} + +func (co *contextObserver) Sync() error { + return nil +} diff --git a/vendor/modules.txt b/vendor/modules.txt index a678f105c..ac3ac554d 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -964,6 +964,7 @@ go.uber.org/zap/internal/exit go.uber.org/zap/internal/pool go.uber.org/zap/internal/stacktrace go.uber.org/zap/zapcore +go.uber.org/zap/zaptest/observer # go.yaml.in/yaml/v3 v3.0.4 ## explicit; go 1.16 go.yaml.in/yaml/v3 From 35a2f7157d45c095bf11de2d5b17a1854968a1d7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Juli=C3=A1n=20Gonz=C3=A1lez?= Date: Mon, 10 Aug 2026 16:35:40 -0700 Subject: [PATCH 3/8] Poll indeterminate seam statuses and capture poll pacing at registration An indeterminate status at the invoke seam was failed immediately while the same status arriving from a poll got the three-strike tolerance; both now take the tolerance path. A settled seam status still resolves like a terminal poll, an unresolved claim without an id keeps the fire-and-forget pass-through, and the status helper shrinks to the two values its call sites pass. Poll pacing moves from package variables into a small struct captured at registration: the detached poll goroutine read the cap on every iteration, so test overrides of the globals were a latent data race. Tests now pass short intervals directly, and a capturing registry drives the wrapped handler under a caller-owned timeout cause to cover the context exit, asserting the budget's cause survives the wrapper. Co-Authored-By: Claude Fable 5 --- pkg/connectorbuilder/actions.go | 59 ++++++++++------- pkg/connectorbuilder/actions_legacy_test.go | 73 ++++++++++++++------- 2 files changed, 85 insertions(+), 47 deletions(-) diff --git a/pkg/connectorbuilder/actions.go b/pkg/connectorbuilder/actions.go index aed89acb1..e34541435 100644 --- a/pkg/connectorbuilder/actions.go +++ b/pkg/connectorbuilder/actions.go @@ -23,13 +23,20 @@ const ( maxConsecutiveStatusErrors = 3 ) -// The legacy status poll starts fast and backs off to a cap so a slow -// action doesn't drain a remote manager's rate-limit budget. Variables so -// tests can drive the loop without real-time waits. -var ( - initialStatusPollInterval = time.Second - maxStatusPollInterval = 30 * time.Second -) +// legacyPollIntervals paces the legacy status poll: it starts fast and backs +// off to a cap so a slow action doesn't drain a remote manager's rate-limit +// budget. Captured at registration, so tests can drive the loop without +// real-time waits and the detached poll goroutine never reads shared +// mutable state. +type legacyPollIntervals struct { + initial time.Duration + max time.Duration +} + +var defaultLegacyPollIntervals = legacyPollIntervals{ + initial: time.Second, + max: 30 * time.Second, +} // ActionManager defines the interface for managing actions in the connector builder. // This is the internal interface used by the builder for dispatch. @@ -228,7 +235,13 @@ func (b *builder) GetActionStatus(ctx context.Context, request *v2.GetActionStat } // registerLegacyAction wraps a legacy CustomActionManager action as an ActionHandler and registers it. -func registerLegacyAction(ctx context.Context, registry actions.ActionRegistry, schema *v2.BatonActionSchema, legacyManager CustomActionManager) error { +func registerLegacyAction( + ctx context.Context, + registry actions.ActionRegistry, + schema *v2.BatonActionSchema, + legacyManager CustomActionManager, + intervals legacyPollIntervals, +) error { handler := func(ctx context.Context, args *structpb.Struct) (*structpb.Struct, annotations.Annotations, error) { // The inner call keeps the detached handler context; its one-hour // deadline is the execution backstop for however long the legacy @@ -245,19 +258,23 @@ func registerLegacyAction(ctx context.Context, registry actions.ActionRegistry, return resp, annos, nil } - // A terminal status at the invoke seam resolves like a terminal + // A settled status at the invoke seam resolves like a terminal // poll. The SDK's own manager reports handler failures in-band as // FAILED with a nil error, so this must not resolve as success. - if !isInFlightActionStatus(actionStatus) { + if actionStatus == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE || actionStatus == v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED { return resp, annos, legacyStatusErr(schema.GetName(), actionStatus) } - // An in-flight claim without an id cannot be polled; resolve with + // An unresolved claim without an id cannot be polled; resolve with // the response, matching the old fire-and-forget behavior. if id == "" { return resp, annos, nil } + // In-flight and indeterminate statuses alike are polled: an + // indeterminate answer gets the same tolerance here as one arriving + // from a later poll. + // Poll to a terminal status so the outer result carries the action's // real outcome. A few consecutive status-check failures are tolerated: // one flaky remote lookup must not convert a succeeding action into a @@ -265,7 +282,7 @@ func registerLegacyAction(ctx context.Context, registry actions.ActionRegistry, // drain a remote manager's rate-limit budget. l := ctxzap.Extract(ctx) statusErrs := 0 - interval := initialStatusPollInterval + interval := intervals.initial timer := time.NewTimer(interval) defer timer.Stop() for { @@ -274,7 +291,7 @@ func registerLegacyAction(ctx context.Context, registry actions.ActionRegistry, return resp, annos, fmt.Errorf("legacy action %q did not reach a terminal status: %w", schema.GetName(), context.Cause(ctx)) case <-timer.C: } - interval = min(interval*2, maxStatusPollInterval) + interval = min(interval*2, intervals.max) timer.Reset(interval) st, _, pollResp, pollAnnos, err := legacyManager.GetActionStatus(ctx, id) @@ -318,17 +335,13 @@ func registerLegacyAction(ctx context.Context, registry actions.ActionRegistry, return registry.Register(ctx, schema, handler) } -// legacyStatusErr maps a settled legacy status to the outer handler error: -// COMPLETE resolves clean, anything else fails the action. +// legacyStatusErr maps a settled legacy status — COMPLETE or FAILED, the +// only values both call sites pass — to the outer handler error. func legacyStatusErr(name string, st v2.BatonActionStatus) error { - switch st { - case v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE: + if st == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE { return nil - case v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED: - return fmt.Errorf("legacy action %q failed", name) - default: - return fmt.Errorf("legacy action %q returned unexpected status %s", name, st.String()) } + return fmt.Errorf("legacy action %q failed", name) } func isInFlightActionStatus(s v2.BatonActionStatus) bool { @@ -345,7 +358,7 @@ func (b *builder) addActionManager(ctx context.Context, in interface{}, registry return fmt.Errorf("error listing schemas from custom action manager: %w", err) } for _, schema := range schemas { - if err := registerLegacyAction(ctx, registry, schema, customManager); err != nil { + if err := registerLegacyAction(ctx, registry, schema, customManager, defaultLegacyPollIntervals); err != nil { return fmt.Errorf("error registering legacy action %s: %w", schema.GetName(), err) } } @@ -366,7 +379,7 @@ func (b *builder) addActionManager(ctx context.Context, in interface{}, registry return fmt.Errorf("error listing schemas from custom action manager: %w", err) } for _, schema := range schemas { - if err := registerLegacyAction(ctx, registry, schema, customManager); err != nil { + if err := registerLegacyAction(ctx, registry, schema, customManager, defaultLegacyPollIntervals); err != nil { return fmt.Errorf("error registering legacy action %s: %w", schema.GetName(), err) } } diff --git a/pkg/connectorbuilder/actions_legacy_test.go b/pkg/connectorbuilder/actions_legacy_test.go index bed6ccf58..c04b247c4 100644 --- a/pkg/connectorbuilder/actions_legacy_test.go +++ b/pkg/connectorbuilder/actions_legacy_test.go @@ -2,7 +2,9 @@ package connectorbuilder import ( "context" + "errors" "fmt" + "slices" "sync/atomic" "testing" "time" @@ -54,7 +56,7 @@ func TestRegisterLegacyActionThirdPartyManagerKeepsHandlerContext(t *testing.T) } m := actions.NewActionManager(ctx) - require.NoError(t, registerLegacyAction(ctx, m, legacy.schema, legacy)) + require.NoError(t, registerLegacyAction(ctx, m, legacy.schema, legacy, shortPollIntervals)) _, _, _, _, err := m.InvokeAction(ctx, "legacy_action", "", &structpb.Struct{}) require.NoError(t, err) @@ -73,7 +75,6 @@ func TestRegisterLegacyActionThirdPartyManagerKeepsHandlerContext(t *testing.T) // finishes: the outer action must stay RUNNING at its own inline wait and // resolve later with the real response, never as an empty completion. func TestRegisterLegacyActionTracksInnerManagerToCompletion(t *testing.T) { - shortStatusPolls(t) ctx := t.Context() schema := v2.BatonActionSchema_builder{Name: "inner_action"}.Build() @@ -87,7 +88,7 @@ func TestRegisterLegacyActionTracksInnerManagerToCompletion(t *testing.T) { })) outer := actions.NewActionManager(ctx) - require.NoError(t, registerLegacyAction(ctx, outer, schema, inner)) + require.NoError(t, registerLegacyAction(ctx, outer, schema, inner, shortPollIntervals)) outerID, outerStatus, outerRv, _, err := outer.InvokeAction(ctx, "inner_action", "", &structpb.Struct{}) require.NoError(t, err) @@ -137,7 +138,6 @@ func (f *fakeAsyncThirdPartyActionManager) GetActionStatus(_ context.Context, id // terminal one with the action id it returned, riding through a transient // status-check failure. func TestRegisterLegacyActionPollsAsyncThirdPartyManager(t *testing.T) { - shortStatusPolls(t) ctx := t.Context() rv, err := structpb.NewStruct(map[string]any{"done": true}) @@ -149,7 +149,7 @@ func TestRegisterLegacyActionPollsAsyncThirdPartyManager(t *testing.T) { } outer := actions.NewActionManager(ctx) - require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy)) + require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy, shortPollIntervals)) outerID, _, _, _, err := outer.InvokeAction(ctx, "async_action", "", &structpb.Struct{}) require.NoError(t, err) @@ -166,7 +166,6 @@ func TestRegisterLegacyActionPollsAsyncThirdPartyManager(t *testing.T) { // A legacy action that polls to FAILED must mark the outer action FAILED, // never resolve it as a success. func TestRegisterLegacyActionPolledFailureFailsOuterAction(t *testing.T) { - shortStatusPolls(t) ctx := t.Context() legacy := &fakeAsyncThirdPartyActionManager{ @@ -175,7 +174,7 @@ func TestRegisterLegacyActionPolledFailureFailsOuterAction(t *testing.T) { } outer := actions.NewActionManager(ctx) - require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy)) + require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy, shortPollIntervals)) outerID, _, _, _, err := outer.InvokeAction(ctx, "failing_action", "", &structpb.Struct{}) require.NoError(t, err) @@ -223,7 +222,7 @@ func TestRegisterLegacyActionSyncManagerWithoutStatusResolves(t *testing.T) { } outer := actions.NewActionManager(ctx) - require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy)) + require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy, shortPollIntervals)) _, outerStatus, outerRv, _, err := outer.InvokeAction(ctx, "sync_action", "", &structpb.Struct{}) require.NoError(t, err) @@ -232,16 +231,9 @@ func TestRegisterLegacyActionSyncManagerWithoutStatusResolves(t *testing.T) { require.Equal(t, int32(0), legacy.statusCalls.Load()) } -// shortStatusPolls drives the poll loop in milliseconds so outcome tables +// shortPollIntervals drives the poll loop in milliseconds so outcome tables // don't add real wall time to the suite. -func shortStatusPolls(t *testing.T) { - t.Helper() - origInitial, origMax := initialStatusPollInterval, maxStatusPollInterval - initialStatusPollInterval, maxStatusPollInterval = time.Millisecond, 4*time.Millisecond - t.Cleanup(func() { - initialStatusPollInterval, maxStatusPollInterval = origInitial, origMax - }) -} +var shortPollIntervals = legacyPollIntervals{initial: time.Millisecond, max: 4 * time.Millisecond} type scriptedPollResult struct { status v2.BatonActionStatus @@ -289,8 +281,6 @@ func (f *scriptedLegacyActionManager) GetActionStatus(_ context.Context, _ strin // poll loop tolerates transient lookup errors and indeterminate statuses up // to the shared threshold before failing closed. func TestRegisterLegacyActionSeamAndPollOutcomes(t *testing.T) { - shortStatusPolls(t) - const ( unspecified = v2.BatonActionStatus_BATON_ACTION_STATUS_UNSPECIFIED unknown = v2.BatonActionStatus_BATON_ACTION_STATUS_UNKNOWN @@ -309,7 +299,9 @@ func TestRegisterLegacyActionSeamAndPollOutcomes(t *testing.T) { wantErrIn string }{ {"in-band failure at the invoke seam fails", "id-1", failed, nil, failed, "failed"}, - {"unexpected status at the invoke seam fails", "id-1", unknown, nil, failed, "unexpected status"}, + {"indeterminate seam status polls to an outcome", "id-1", unknown, + []scriptedPollResult{{status: complete}}, complete, ""}, + {"indeterminate seam status without an id resolves as before", "", unknown, nil, complete, ""}, {"unspecified status resolves as before", "id-1", unspecified, nil, complete, ""}, {"in-flight without an id resolves as before", "", running, nil, complete, ""}, {"threshold consecutive lookup errors fail closed", "id-1", running, @@ -334,7 +326,7 @@ func TestRegisterLegacyActionSeamAndPollOutcomes(t *testing.T) { polls: tc.polls, } outer := actions.NewActionManager(ctx) - require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy)) + require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy, shortPollIntervals)) _, st, gotRv, _, err := outer.InvokeAction(ctx, "scripted_action", "", &structpb.Struct{}) require.NoError(t, err) @@ -351,8 +343,6 @@ func TestRegisterLegacyActionSeamAndPollOutcomes(t *testing.T) { // The poll loop's warnings must reach the caller's logger: the detached // handler context carries it across the goroutine boundary. func TestLegacyPollWarningsReachCallerLogger(t *testing.T) { - shortStatusPolls(t) - core, observed := observer.New(zap.WarnLevel) ctx := ctxzap.ToContext(t.Context(), zap.New(core)) @@ -369,7 +359,7 @@ func TestLegacyPollWarningsReachCallerLogger(t *testing.T) { }, } outer := actions.NewActionManager(ctx) - require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy)) + require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy, shortPollIntervals)) _, st, _, _, err := outer.InvokeAction(ctx, "warned_action", "", &structpb.Struct{}) require.NoError(t, err) @@ -378,3 +368,38 @@ func TestLegacyPollWarningsReachCallerLogger(t *testing.T) { return observed.FilterMessage("legacy action status check failed, retrying").Len() == 1 }, time.Second, 10*time.Millisecond) } + +type capturingRegistry struct { + handler actions.ActionHandler +} + +func (c *capturingRegistry) Register(_ context.Context, _ *v2.BatonActionSchema, handler actions.ActionHandler) error { + c.handler = handler + return nil +} + +func (c *capturingRegistry) RegisterAction(_ context.Context, _ string, _ *v2.BatonActionSchema, handler actions.ActionHandler) error { + c.handler = handler + return nil +} + +// The poll loop's context exit must surface the handler budget's cause, not +// a bare context error. +func TestLegacyPollSurfacesHandlerBudgetCause(t *testing.T) { + legacy := &scriptedLegacyActionManager{ + schema: v2.BatonActionSchema_builder{Name: "budgeted_action"}.Build(), + invokeID: "id-1", + invokeStatus: v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING, + polls: slices.Repeat([]scriptedPollResult{{status: v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING}}, 64), + } + reg := &capturingRegistry{} + require.NoError(t, registerLegacyAction(t.Context(), reg, legacy.schema, legacy, shortPollIntervals)) + + cause := errors.New("action handler timed out") + handlerCtx, cancel := context.WithTimeoutCause(t.Context(), 25*time.Millisecond, cause) + defer cancel() + + _, _, err := reg.handler(handlerCtx, &structpb.Struct{}) + require.ErrorIs(t, err, cause) + require.ErrorContains(t, err, "did not reach a terminal status") +} From 72d194af6b26b0b5cfd38efe04c235c65df978d6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Juli=C3=A1n=20Gonz=C3=A1lez?= Date: Mon, 10 Aug 2026 17:02:29 -0700 Subject: [PATCH 4/8] Instrument the payload guard, share the settled-status gate, guard poll pacing The meaningful-payload guard in the poll loop had no test that could fail if it were removed, because every fake returned the same payload for invokes and polls. Scripted polls can now carry their own payload, and a displacement test invokes with one payload, feeds three indeterminate polls carrying another, and asserts the fail-closed exit still returns the invoke's payload. The settled-status check is one predicate now, used at the invoke seam, the poll switch, and the retention guard, so the gates cannot drift; a new terminal enum value joins there or takes the indeterminate path. Zero or negative poll intervals fall back to the defaults instead of busy-looping the status poll. Co-Authored-By: Claude Fable 5 --- pkg/connectorbuilder/actions.go | 18 +++++++++-- pkg/connectorbuilder/actions_legacy_test.go | 36 +++++++++++++++++++++ 2 files changed, 51 insertions(+), 3 deletions(-) diff --git a/pkg/connectorbuilder/actions.go b/pkg/connectorbuilder/actions.go index e34541435..d55773660 100644 --- a/pkg/connectorbuilder/actions.go +++ b/pkg/connectorbuilder/actions.go @@ -242,6 +242,10 @@ func registerLegacyAction( legacyManager CustomActionManager, intervals legacyPollIntervals, ) error { + // A zero or negative interval would busy-loop the status poll. + if intervals.initial <= 0 || intervals.max <= 0 { + intervals = defaultLegacyPollIntervals + } handler := func(ctx context.Context, args *structpb.Struct) (*structpb.Struct, annotations.Annotations, error) { // The inner call keeps the detached handler context; its one-hour // deadline is the execution backstop for however long the legacy @@ -261,7 +265,7 @@ func registerLegacyAction( // A settled status at the invoke seam resolves like a terminal // poll. The SDK's own manager reports handler failures in-band as // FAILED with a nil error, so this must not resolve as success. - if actionStatus == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE || actionStatus == v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED { + if isSettledActionStatus(actionStatus) { return resp, annos, legacyStatusErr(schema.GetName(), actionStatus) } @@ -308,14 +312,14 @@ func registerLegacyAction( } // Keep the last meaningful response for the error exits above; // an indeterminate poll's payload must not replace it. - if pollResp != nil && (isInFlightActionStatus(st) || st == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE || st == v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED) { + if pollResp != nil && (isInFlightActionStatus(st) || isSettledActionStatus(st)) { resp, annos = pollResp, pollAnnos } switch { case isInFlightActionStatus(st): statusErrs = 0 - case st == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE || st == v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED: + case isSettledActionStatus(st): return resp, annos, legacyStatusErr(schema.GetName(), st) default: // An indeterminate status gets the same tolerance as a @@ -348,6 +352,14 @@ func isInFlightActionStatus(s v2.BatonActionStatus) bool { return s == v2.BatonActionStatus_BATON_ACTION_STATUS_PENDING || s == v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING } +// isSettledActionStatus is the single gate deciding which statuses resolve +// immediately, at the invoke seam and from polls alike. A new terminal enum +// value must be added here, or it takes the indeterminate path: polled to +// the tolerance threshold, then failed closed. +func isSettledActionStatus(s v2.BatonActionStatus) bool { + return s == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE || s == v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED +} + // addActionManager handles deprecated CustomActionManager and RegisterActionManagerLimited interfaces // by extracting their actions and registering them into the unified ActionManager. func (b *builder) addActionManager(ctx context.Context, in interface{}, registry actions.ActionRegistry) error { diff --git a/pkg/connectorbuilder/actions_legacy_test.go b/pkg/connectorbuilder/actions_legacy_test.go index c04b247c4..c226f1e60 100644 --- a/pkg/connectorbuilder/actions_legacy_test.go +++ b/pkg/connectorbuilder/actions_legacy_test.go @@ -238,6 +238,7 @@ var shortPollIntervals = legacyPollIntervals{initial: time.Millisecond, max: 4 * type scriptedPollResult struct { status v2.BatonActionStatus err error + resp *structpb.Struct } type scriptedLegacyActionManager struct { @@ -272,6 +273,9 @@ func (f *scriptedLegacyActionManager) GetActionStatus(_ context.Context, _ strin if p.err != nil { return v2.BatonActionStatus_BATON_ACTION_STATUS_UNKNOWN, "", nil, nil, p.err } + if p.resp != nil { + return p.status, "scripted", p.resp, nil, nil + } return p.status, "scripted", f.invokeRv, nil, nil } @@ -403,3 +407,35 @@ func TestLegacyPollSurfacesHandlerBudgetCause(t *testing.T) { require.ErrorIs(t, err, cause) require.ErrorContains(t, err, "did not reach a terminal status") } + +// Removing the meaningful-payload guard in the poll loop must fail this +// test: an indeterminate poll's payload must not displace the invoke +// response that the fail-closed exit returns. +func TestLegacyIndeterminatePollPayloadDoesNotDisplaceResponse(t *testing.T) { + ctx := t.Context() + + invokePayload, err := structpb.NewStruct(map[string]any{"from": "invoke"}) + require.NoError(t, err) + pollPayload, err := structpb.NewStruct(map[string]any{"from": "poll"}) + require.NoError(t, err) + + indeterminate := scriptedPollResult{ + status: v2.BatonActionStatus_BATON_ACTION_STATUS_UNSPECIFIED, + resp: pollPayload, + } + legacy := &scriptedLegacyActionManager{ + schema: v2.BatonActionSchema_builder{Name: "displacing_action"}.Build(), + invokeID: "id-1", + invokeStatus: v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING, + invokeRv: invokePayload, + polls: []scriptedPollResult{indeterminate, indeterminate, indeterminate}, + } + outer := actions.NewActionManager(ctx) + require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy, shortPollIntervals)) + + _, st, gotRv, _, err := outer.InvokeAction(ctx, "displacing_action", "", &structpb.Struct{}) + require.NoError(t, err) + require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED, st) + require.Equal(t, "invoke", gotRv.Fields["from"].GetStringValue()) + require.Contains(t, gotRv.Fields["error"].GetStringValue(), "unexpected status") +} From fc2d24dc7568f20954bb4cec263982768718b77c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Juli=C3=A1n=20Gonz=C3=A1lez?= Date: Mon, 10 Aug 2026 17:14:38 -0700 Subject: [PATCH 5/8] Instrument the poll-pacing fallback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The zero-interval guard had no test that could fail if it were removed — the same gap the payload-displacement instrument closed one commit earlier. A zero-valued pacing struct now proves the fallback: the handler resolves terminally with exactly one poll, and that poll must wait the defaults' initial tick rather than firing immediately, which is the observable difference between the guard and a busy loop. A poll count alone cannot discriminate, since a spinning loop with a single scripted poll also polls exactly once. The guard's comment also covers the inverted-pair case, which needs no normalization: the cap applies from the second tick. Co-Authored-By: Claude Fable 5 --- pkg/connectorbuilder/actions.go | 3 ++- pkg/connectorbuilder/actions_legacy_test.go | 25 +++++++++++++++++++++ 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/pkg/connectorbuilder/actions.go b/pkg/connectorbuilder/actions.go index d55773660..ae6d4fd33 100644 --- a/pkg/connectorbuilder/actions.go +++ b/pkg/connectorbuilder/actions.go @@ -242,7 +242,8 @@ func registerLegacyAction( legacyManager CustomActionManager, intervals legacyPollIntervals, ) error { - // A zero or negative interval would busy-loop the status poll. + // A zero or negative interval would busy-loop the status poll. An + // inverted pair needs no guard: the cap applies from the second tick. if intervals.initial <= 0 || intervals.max <= 0 { intervals = defaultLegacyPollIntervals } diff --git a/pkg/connectorbuilder/actions_legacy_test.go b/pkg/connectorbuilder/actions_legacy_test.go index c226f1e60..129d341b7 100644 --- a/pkg/connectorbuilder/actions_legacy_test.go +++ b/pkg/connectorbuilder/actions_legacy_test.go @@ -439,3 +439,28 @@ func TestLegacyIndeterminatePollPayloadDoesNotDisplaceResponse(t *testing.T) { require.Equal(t, "invoke", gotRv.Fields["from"].GetStringValue()) require.Contains(t, gotRv.Fields["error"].GetStringValue(), "unexpected status") } + +// Removing the interval fallback must fail this test: with a zero-valued +// pacing struct the first poll must still wait the defaults' initial tick +// rather than firing immediately (and then busy-looping on a zero cap). +func TestZeroPollIntervalsFallBackToDefaults(t *testing.T) { + rv, err := structpb.NewStruct(map[string]any{"done": true}) + require.NoError(t, err) + legacy := &scriptedLegacyActionManager{ + schema: v2.BatonActionSchema_builder{Name: "unpaced_action"}.Build(), + invokeID: "id-1", + invokeStatus: v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING, + invokeRv: rv, + polls: []scriptedPollResult{{status: v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE}}, + } + reg := &capturingRegistry{} + require.NoError(t, registerLegacyAction(t.Context(), reg, legacy.schema, legacy, legacyPollIntervals{})) + + start := time.Now() + _, _, err = reg.handler(t.Context(), &structpb.Struct{}) + elapsed := time.Since(start) + + require.NoError(t, err) + require.EqualValues(t, 1, legacy.pollCalls.Load()) + require.GreaterOrEqual(t, elapsed, 900*time.Millisecond) +} From e29fad8996e581fe287e1b53158df1b1fa6bc7fc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Juli=C3=A1n=20Gonz=C3=A1lez?= Date: Mon, 10 Aug 2026 17:29:25 -0700 Subject: [PATCH 6/8] Cover both halves of the poll-pacing fallback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The fallback test only passed a zero-valued pacing struct, which trips the initial-interval check on its own — weakening the guard to ignore the cap left the suite green while a live initial with a zero cap busy-loops from the second tick. The test is now a table whose second case passes exactly that shape; because the fallback replaces the whole struct, the same one-poll elapsed lower bound discriminates both halves: a guarded poll waits the defaults' initial tick, an unguarded one fires at the un-defaulted initial. Co-Authored-By: Claude Fable 5 --- pkg/connectorbuilder/actions_legacy_test.go | 53 +++++++++++++-------- 1 file changed, 33 insertions(+), 20 deletions(-) diff --git a/pkg/connectorbuilder/actions_legacy_test.go b/pkg/connectorbuilder/actions_legacy_test.go index 129d341b7..f1fdb1005 100644 --- a/pkg/connectorbuilder/actions_legacy_test.go +++ b/pkg/connectorbuilder/actions_legacy_test.go @@ -440,27 +440,40 @@ func TestLegacyIndeterminatePollPayloadDoesNotDisplaceResponse(t *testing.T) { require.Contains(t, gotRv.Fields["error"].GetStringValue(), "unexpected status") } -// Removing the interval fallback must fail this test: with a zero-valued -// pacing struct the first poll must still wait the defaults' initial tick -// rather than firing immediately (and then busy-looping on a zero cap). -func TestZeroPollIntervalsFallBackToDefaults(t *testing.T) { - rv, err := structpb.NewStruct(map[string]any{"done": true}) - require.NoError(t, err) - legacy := &scriptedLegacyActionManager{ - schema: v2.BatonActionSchema_builder{Name: "unpaced_action"}.Build(), - invokeID: "id-1", - invokeStatus: v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING, - invokeRv: rv, - polls: []scriptedPollResult{{status: v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE}}, +// Weakening either half of the interval fallback must fail a case here: the +// fallback replaces the whole pacing struct, so even a non-default initial +// paired with a zero cap (which would busy-loop from the second tick) must +// be re-paced to the defaults' initial tick, which is what the elapsed +// lower bound proves. +func TestNonPositivePollIntervalsFallBackToDefaults(t *testing.T) { + cases := []struct { + name string + intervals legacyPollIntervals + }{ + {"both zero", legacyPollIntervals{}}, + {"zero cap with a live initial", legacyPollIntervals{initial: 50 * time.Millisecond}}, } - reg := &capturingRegistry{} - require.NoError(t, registerLegacyAction(t.Context(), reg, legacy.schema, legacy, legacyPollIntervals{})) + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + rv, err := structpb.NewStruct(map[string]any{"done": true}) + require.NoError(t, err) + legacy := &scriptedLegacyActionManager{ + schema: v2.BatonActionSchema_builder{Name: "unpaced_action"}.Build(), + invokeID: "id-1", + invokeStatus: v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING, + invokeRv: rv, + polls: []scriptedPollResult{{status: v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE}}, + } + reg := &capturingRegistry{} + require.NoError(t, registerLegacyAction(t.Context(), reg, legacy.schema, legacy, tc.intervals)) - start := time.Now() - _, _, err = reg.handler(t.Context(), &structpb.Struct{}) - elapsed := time.Since(start) + start := time.Now() + _, _, err = reg.handler(t.Context(), &structpb.Struct{}) + elapsed := time.Since(start) - require.NoError(t, err) - require.EqualValues(t, 1, legacy.pollCalls.Load()) - require.GreaterOrEqual(t, elapsed, 900*time.Millisecond) + require.NoError(t, err) + require.EqualValues(t, 1, legacy.pollCalls.Load()) + require.GreaterOrEqual(t, elapsed, 900*time.Millisecond) + }) + } } From 805639e16077c6262f2a76ea3655cda6138ed79b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Juli=C3=A1n=20Gonz=C3=A1lez?= Date: Mon, 10 Aug 2026 17:41:43 -0700 Subject: [PATCH 7/8] Carry the inner error through legacy failure reporting The outer error replaces the response's error field, so the generic 'legacy action failed' message was destroying the inner manager's real failure text at exactly the moment it should surface. legacyStatusErr now folds the response's reported error into the outer message, at the invoke seam and from polls alike. The lookup-error threshold exit also wraps with the action name like the loop's other exits, instead of returning a bare context error, and the threshold test asserts on the wrapper text so the wrap itself is pinned. Co-Authored-By: Claude Fable 5 --- pkg/connectorbuilder/actions.go | 15 ++++--- pkg/connectorbuilder/actions_legacy_test.go | 47 ++++++++++++++++++++- 2 files changed, 56 insertions(+), 6 deletions(-) diff --git a/pkg/connectorbuilder/actions.go b/pkg/connectorbuilder/actions.go index ae6d4fd33..956e00be6 100644 --- a/pkg/connectorbuilder/actions.go +++ b/pkg/connectorbuilder/actions.go @@ -267,7 +267,7 @@ func registerLegacyAction( // poll. The SDK's own manager reports handler failures in-band as // FAILED with a nil error, so this must not resolve as success. if isSettledActionStatus(actionStatus) { - return resp, annos, legacyStatusErr(schema.GetName(), actionStatus) + return resp, annos, legacyStatusErr(schema.GetName(), actionStatus, resp) } // An unresolved claim without an id cannot be polled; resolve with @@ -303,7 +303,7 @@ func registerLegacyAction( if err != nil { statusErrs++ if statusErrs >= maxConsecutiveStatusErrors { - return resp, annos, err + return resp, annos, fmt.Errorf("legacy action %q status lookup failed: %w", schema.GetName(), err) } l.Warn("legacy action status check failed, retrying", zap.String("action", schema.GetName()), @@ -321,7 +321,7 @@ func registerLegacyAction( case isInFlightActionStatus(st): statusErrs = 0 case isSettledActionStatus(st): - return resp, annos, legacyStatusErr(schema.GetName(), st) + return resp, annos, legacyStatusErr(schema.GetName(), st, resp) default: // An indeterminate status gets the same tolerance as a // lookup error: transient anomalies recover, persistent @@ -341,11 +341,16 @@ func registerLegacyAction( } // legacyStatusErr maps a settled legacy status — COMPLETE or FAILED, the -// only values both call sites pass — to the outer handler error. -func legacyStatusErr(name string, st v2.BatonActionStatus) error { +// only values both call sites pass — to the outer handler error, carrying +// the inner manager's reported error message when the response has one, +// since the outer error replaces the response's error field. +func legacyStatusErr(name string, st v2.BatonActionStatus, resp *structpb.Struct) error { if st == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE { return nil } + if inner := resp.GetFields()["error"].GetStringValue(); inner != "" { + return fmt.Errorf("legacy action %q failed: %s", name, inner) + } return fmt.Errorf("legacy action %q failed", name) } diff --git a/pkg/connectorbuilder/actions_legacy_test.go b/pkg/connectorbuilder/actions_legacy_test.go index f1fdb1005..d81802b74 100644 --- a/pkg/connectorbuilder/actions_legacy_test.go +++ b/pkg/connectorbuilder/actions_legacy_test.go @@ -309,7 +309,7 @@ func TestRegisterLegacyActionSeamAndPollOutcomes(t *testing.T) { {"unspecified status resolves as before", "id-1", unspecified, nil, complete, ""}, {"in-flight without an id resolves as before", "", running, nil, complete, ""}, {"threshold consecutive lookup errors fail closed", "id-1", running, - []scriptedPollResult{lookupErr, lookupErr, lookupErr}, failed, "deadline"}, + []scriptedPollResult{lookupErr, lookupErr, lookupErr}, failed, "status lookup failed"}, {"lookup errors under the threshold recover", "id-1", running, []scriptedPollResult{lookupErr, lookupErr, {status: complete}}, complete, ""}, {"indeterminate polls under the threshold recover", "id-1", running, @@ -477,3 +477,48 @@ func TestNonPositivePollIntervalsFallBackToDefaults(t *testing.T) { }) } } + +// The inner manager's reported error must survive into the outer error, +// which replaces the response's error field: at the seam and from a poll +// alike, the failed outcome carries the inner message, not just the +// generic wrapper text. +func TestLegacyFailureCarriesInnerError(t *testing.T) { + failedRv, err := structpb.NewStruct(map[string]any{"error": "boom from inner"}) + require.NoError(t, err) + + t.Run("at the invoke seam", func(t *testing.T) { + ctx := t.Context() + legacy := &scriptedLegacyActionManager{ + schema: v2.BatonActionSchema_builder{Name: "inner_error_action"}.Build(), + invokeID: "id-1", + invokeStatus: v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED, + invokeRv: failedRv, + } + outer := actions.NewActionManager(ctx) + require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy, shortPollIntervals)) + + _, st, gotRv, _, err := outer.InvokeAction(ctx, "inner_error_action", "", &structpb.Struct{}) + require.NoError(t, err) + require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED, st) + require.Contains(t, gotRv.Fields["error"].GetStringValue(), "boom from inner") + }) + + t.Run("from a poll", func(t *testing.T) { + ctx := t.Context() + legacy := &scriptedLegacyActionManager{ + schema: v2.BatonActionSchema_builder{Name: "inner_error_action"}.Build(), + invokeID: "id-1", + invokeStatus: v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING, + polls: []scriptedPollResult{ + {status: v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED, resp: failedRv}, + }, + } + outer := actions.NewActionManager(ctx) + require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy, shortPollIntervals)) + + _, st, gotRv, _, err := outer.InvokeAction(ctx, "inner_error_action", "", &structpb.Struct{}) + require.NoError(t, err) + require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED, st) + require.Contains(t, gotRv.Fields["error"].GetStringValue(), "boom from inner") + }) +} From d7d74095d04c89e2f24634254f3221232e4c1321 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Juli=C3=A1n=20Gonz=C3=A1lez?= Date: Mon, 10 Aug 2026 17:51:49 -0700 Subject: [PATCH 8/8] Report the settling poll's payload as the failure cause MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The settled exit read the retained response for the inner error message, so an in-flight poll's snapshot could be reported as the failure cause when the poll that settled to FAILED carried no payload of its own. The error-message source is now the settling poll's payload — nil falls back to the generic message — while the retained response remains the returned response value. Co-Authored-By: Claude Fable 5 --- pkg/connectorbuilder/actions.go | 8 ++++--- pkg/connectorbuilder/actions_legacy_test.go | 26 +++++++++++++++++++++ 2 files changed, 31 insertions(+), 3 deletions(-) diff --git a/pkg/connectorbuilder/actions.go b/pkg/connectorbuilder/actions.go index 956e00be6..8f0038c6b 100644 --- a/pkg/connectorbuilder/actions.go +++ b/pkg/connectorbuilder/actions.go @@ -321,7 +321,9 @@ func registerLegacyAction( case isInFlightActionStatus(st): statusErrs = 0 case isSettledActionStatus(st): - return resp, annos, legacyStatusErr(schema.GetName(), st, resp) + // The settling poll's own payload is the failure account; + // resp may hold an older in-flight snapshot. + return resp, annos, legacyStatusErr(schema.GetName(), st, pollResp) default: // An indeterminate status gets the same tolerance as a // lookup error: transient anomalies recover, persistent @@ -342,8 +344,8 @@ func registerLegacyAction( // legacyStatusErr maps a settled legacy status — COMPLETE or FAILED, the // only values both call sites pass — to the outer handler error, carrying -// the inner manager's reported error message when the response has one, -// since the outer error replaces the response's error field. +// the error message the settling response reports, since the outer error +// replaces the response's error field. func legacyStatusErr(name string, st v2.BatonActionStatus, resp *structpb.Struct) error { if st == v2.BatonActionStatus_BATON_ACTION_STATUS_COMPLETE { return nil diff --git a/pkg/connectorbuilder/actions_legacy_test.go b/pkg/connectorbuilder/actions_legacy_test.go index d81802b74..2e6170b4d 100644 --- a/pkg/connectorbuilder/actions_legacy_test.go +++ b/pkg/connectorbuilder/actions_legacy_test.go @@ -522,3 +522,29 @@ func TestLegacyFailureCarriesInnerError(t *testing.T) { require.Contains(t, gotRv.Fields["error"].GetStringValue(), "boom from inner") }) } + +// A stale in-flight payload must not be reported as the failure cause when +// the settling poll itself carries no payload. +func TestLegacyStaleInFlightPayloadIsNotTheFailureCause(t *testing.T) { + ctx := t.Context() + + staleRv, err := structpb.NewStruct(map[string]any{"error": "poll1-snapshot"}) + require.NoError(t, err) + legacy := &scriptedLegacyActionManager{ + schema: v2.BatonActionSchema_builder{Name: "stale_error_action"}.Build(), + invokeID: "id-1", + invokeStatus: v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING, + polls: []scriptedPollResult{ + {status: v2.BatonActionStatus_BATON_ACTION_STATUS_RUNNING, resp: staleRv}, + {status: v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED}, + }, + } + outer := actions.NewActionManager(ctx) + require.NoError(t, registerLegacyAction(ctx, outer, legacy.schema, legacy, shortPollIntervals)) + + _, st, gotRv, _, err := outer.InvokeAction(ctx, "stale_error_action", "", &structpb.Struct{}) + require.NoError(t, err) + require.Equal(t, v2.BatonActionStatus_BATON_ACTION_STATUS_FAILED, st) + require.NotContains(t, gotRv.Fields["error"].GetStringValue(), "poll1-snapshot") + require.Contains(t, gotRv.Fields["error"].GetStringValue(), "failed") +}