From a609c33e0bd8d13cd6115b2281b42b7af13bf00d Mon Sep 17 00:00:00 2001 From: Matthew Scobell Date: Thu, 13 Aug 2026 18:24:41 -0400 Subject: [PATCH 01/11] fix: hot-load file changes on every sync in long-running services MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The parsed-file cache was built exactly once at process startup inside ResourceSyncers(), so long-lived services served stale data until restart. Validate() — which the SDK runs at the start of every sync — now rebuilds the cache from the current file contents and publishes it atomically to a holder shared by all resource builders, restoring the hot-load behavior that regressed in #40. - Generation-stamped page tokens: a listing resumed across a cache swap restarts instead of silently skipping/duplicating rows; legacy bare numeric tokens are still honored so in-flight syncs survive upgrades - Nil-cache guards in List/Entitlements/Grants replace potential panics - An invalid file edit now fails the sync loudly while the last successfully loaded data keeps serving - nolint:staticcheck on the deprecated trait options: they populate both trait- and resource-level fields, so migrating to the WithResource* replacements would drop trait-level fields from sync output - Regression tests enforce the hot-load contract; README documents it Co-Authored-By: Claude Fable 5 --- README.md | 6 + pkg/connector/connector.go | 141 +++++++++++++++++------- pkg/connector/external_grants_test.go | 14 +-- pkg/connector/hot_reload_test.go | 151 ++++++++++++++++++++++++++ pkg/connector/resources.go | 104 +++++++++++++----- pkg/connector/resources_test.go | 68 ++++++++++-- 6 files changed, 399 insertions(+), 85 deletions(-) create mode 100644 pkg/connector/hot_reload_test.go diff --git a/README.md b/README.md index c266f8d1..763d523a 100644 --- a/README.md +++ b/README.md @@ -39,6 +39,12 @@ baton-file -i data.csv baton-file -i data.xlsx ``` +## Hot-Load (File Change Detection) + +When baton-file runs as a long-lived service, edits to the input file are picked up automatically at the start of the next sync cycle — no restart required. This applies to the **data** sections: users, resources, entitlements, grants, and inheritance mappings. **Schema changes** — introducing a brand-new resource type — still require a service restart, because the SDK registers resource types once at process startup. If the edited file fails validation, the sync fails with the validation error and the connector keeps serving the last successfully loaded data. If the file changes while a sync is in flight (for example via health-check revalidation), affected listings restart safely against the new contents instead of resuming into stale page offsets. + +**IMPORTANT (maintainers and AI agents):** this behavior is a deliberate contract, not an accident of implementation. The SDK calls `ResourceSyncers()` exactly once per process, so `Validate()` — which runs at the start of every sync — is the only per-sync hook and is where the file is re-read and the shared cache is republished (see `cacheHolder` in `pkg/connector/connector.go`). Do not move the file load to construction time, capture cache snapshots in builders, or assume the SDK refreshes data between syncs (it does not — that assumption caused a regression once already). `TestHotReload_DataChangesPickedUpBySync` in `pkg/connector/hot_reload_test.go` enforces this contract; if it fails after your change, the change is wrong, not the test. + ## Templates Full templates demonstrate every field and feature. Quickstart templates have the minimum to get a working sync (two users, a group, a role, and direct grants). diff --git a/pkg/connector/connector.go b/pkg/connector/connector.go index 13524cf7..be83d5e9 100644 --- a/pkg/connector/connector.go +++ b/pkg/connector/connector.go @@ -2,10 +2,13 @@ package connector import ( "context" + "crypto/sha256" + "encoding/hex" "fmt" "os" "sort" "strings" + "sync/atomic" "github.com/conductorone/baton-file/pkg/client" v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" @@ -24,9 +27,49 @@ import ( type FileConnector struct { inputFilePath string - validatedData *client.LoadedData + cache cacheHolder } +// cacheHolder shares the live syncCache between the FileConnector and every +// resourceBuilder, so the cache built from the current file contents can be +// swapped in atomically on each sync. +// +// IMPORTANT — hot-load contract (do not break): +// The SDK constructs the connector and calls ResourceSyncers() exactly ONCE +// per process, before its task loop starts. In a long-running service the +// only per-sync hook this connector gets is Validate(), so Validate() is +// where the input file is re-read and the fresh syncCache is published here. +// Data-section changes to the file MUST be picked up on the next sync without +// a restart; only schema changes (a new resource type) require one. Do NOT +// move the file load back to construction time, capture a *syncCache snapshot +// in a builder, or assume the SDK refreshes anything between syncs (it does +// not — that assumption caused the original hot-load regression in PR #40). +// TestHotReload_DataChangesPickedUpBySync enforces this contract. +// +// This is deliberate instance state, a documented deviation from the skills' +// stateless-connector rule: that rule assumes a remote API as the source of +// truth, whereas here the file IS the source of truth and is re-read on every +// sync, so a cold start is identical to a warm one. The swap normally happens +// at sync start (Validate), but the SDK's health-check endpoint also calls +// Validate and can swap mid-sync; page tokens are therefore stamped with the +// cache generation (syncCache.gen) so an in-flight listing restarts instead +// of resuming into a different generation's offsets. During a swap both old +// and new caches are briefly live (~2x peak memory for the dataset). +type cacheHolder struct { + p atomic.Pointer[syncCache] +} + +// load is nil-receiver-safe because StaticCapabilitiesConnector builds +// resourceBuilders without a holder. +func (h *cacheHolder) load() *syncCache { + if h == nil { + return nil + } + return h.p.Load() +} + +func (h *cacheHolder) store(cache *syncCache) { h.p.Store(cache) } + func (fc *FileConnector) Close() error { return nil } // StaticCapabilitiesConnector used exclusively by the capabilities sub-command via WithDefaultCapabilitiesConnectorBuilderV2. @@ -57,6 +100,12 @@ func (s *StaticCapabilitiesConnector) ResourceSyncers(_ context.Context) []conne } type syncCache struct { + // gen fingerprints the file contents this cache was built from; it is + // stamped into page tokens so a listing resumed against a different + // generation restarts instead of replaying offsets into changed slices + // (see paginate). Empty for caches built directly in tests. + gen string + resourceTypes map[string]*v2.ResourceType resources map[string]*v2.Resource entitlements map[string]*v2.Entitlement @@ -109,7 +158,35 @@ func (fc *FileConnector) Validate(ctx context.Context) (annotations.Annotations, return nil, fmt.Errorf("baton-file: error accessing input file: %w", err) } - data, err := client.LoadFileData(fc.inputFilePath) + // Re-read, validate, and publish the file's current contents. Validate() + // runs at the start of every sync, so this is the hot-load refresh point + // (see cacheHolder). On error we return without storing, so the + // previously published cache keeps serving last-known-good data. + cache, err := loadValidatedCache(ctx, fc.inputFilePath) + if err != nil { + return nil, err + } + fc.cache.store(cache) + + ctxzap.Extract(ctx).Debug("baton-file: refreshed data from input file", + zap.Int("resource_types", len(cache.resourceTypes)), + zap.Int("resources", len(cache.resources))) + + return nil, nil +} + +// loadValidatedCache reads the input file, runs all cross-record validations, +// and builds the derived sync cache from it. +func loadValidatedCache(ctx context.Context, inputFilePath string) (*syncCache, error) { + // Fingerprint the raw bytes before parsing; a file edit racing between + // the two reads mislabels one build, which at worst restarts a listing. + raw, err := os.ReadFile(inputFilePath) + if err != nil { + return nil, fmt.Errorf("baton-file: failed to read input file: %w", err) + } + sum := sha256.Sum256(raw) + + data, err := client.LoadFileData(inputFilePath) if err != nil { return nil, fmt.Errorf("baton-file: input file is invalid: %w", err) } @@ -134,58 +211,40 @@ func (fc *FileConnector) Validate(ctx context.Context) (annotations.Annotations, return nil, err } - fc.validatedData = data - - return nil, nil + cache, err := newSyncCache(ctx, data) + if err != nil { + return nil, err + } + cache.gen = hex.EncodeToString(sum[:8]) + return cache, nil } func (fc *FileConnector) ResourceSyncers(ctx context.Context) []connectorbuilder.ResourceSyncerV2 { l := ctxzap.Extract(ctx) - // Validate() caches parsed data; reuse it here to avoid a double load. - // Fallback re-loads if Validate() wasn't called. This method's signature - // cannot return an error, so log-and-return-nil is intentional — the - // connector stays alive for the next sync cycle. - loadedData := fc.validatedData - if loadedData == nil { + // IMPORTANT: the SDK calls this exactly ONCE per process, at connector + // construction, to register resource types — never again. Only the schema + // (the set of syncers) is fixed here; the data each builder serves flows + // through the shared fc.cache holder so Validate() can refresh it every + // sync. See cacheHolder for the hot-load contract. + cache := fc.cache.load() + if cache == nil { + // Validate() normally runs first and publishes the cache; this + // fallback covers entry points that skip Validate(). This method's + // signature cannot return an error, so log-and-return-nil is + // intentional — the connector stays alive for the next sync cycle. var err error - loadedData, err = client.LoadFileData(fc.inputFilePath) + cache, err = loadValidatedCache(ctx, fc.inputFilePath) if err != nil { - l.Error("baton-file: failed to load input file", zap.Error(err)) + l.Warn("baton-file: failed to load input file", zap.Error(err)) return nil } - if err := client.ValidateUniqueIDs(loadedData); err != nil { - l.Error("baton-file: validation failed", zap.Error(err)) - return nil - } - if err := client.ValidateTraits(loadedData); err != nil { - l.Error("baton-file: validation failed", zap.Error(err)) - return nil - } - if err := client.ValidateEntitlementFields(loadedData); err != nil { - l.Error("baton-file: validation failed", zap.Error(err)) - return nil - } - if err := client.ValidateSecretFields(loadedData); err != nil { - l.Error("baton-file: validation failed", zap.Error(err)) - return nil - } - if err := client.ValidateReferences(loadedData); err != nil { - l.Error("baton-file: validation failed", zap.Error(err)) - return nil - } - } - fc.validatedData = nil - - cache, err := newSyncCache(ctx, loadedData) - if err != nil { - l.Error("baton-file: failed to build sync cache", zap.Error(err)) - return nil + fc.cache.store(cache) } var syncers []connectorbuilder.ResourceSyncerV2 for _, rt := range cache.resourceTypes { - syncers = append(syncers, &resourceBuilder{cache: cache, resourceType: rt}) + syncers = append(syncers, &resourceBuilder{cache: &fc.cache, resourceType: rt}) } l.Info("baton-file: created resource syncers", zap.Int("count", len(syncers))) diff --git a/pkg/connector/external_grants_test.go b/pkg/connector/external_grants_test.go index a05498cd..e9735743 100644 --- a/pkg/connector/external_grants_test.go +++ b/pkg/connector/external_grants_test.go @@ -50,7 +50,7 @@ func TestGrants_ExternalGrant_AttributeMatch(t *testing.T) { cache, err := newSyncCache(ctx, data) require.NoError(t, err) - b := &resourceBuilder{cache: cache, resourceType: cache.resourceTypes["app"]} + b := testBuilder(cache, cache.resourceTypes["app"]) grants := allGrants(t, b, cache.resources["payroll-app"]) require.Len(t, grants, 1) @@ -77,7 +77,7 @@ func TestGrants_ExternalGrant_MatchAll(t *testing.T) { cache, err := newSyncCache(ctx, data) require.NoError(t, err) - b := &resourceBuilder{cache: cache, resourceType: cache.resourceTypes["app"]} + b := testBuilder(cache, cache.resourceTypes["app"]) grants := allGrants(t, b, cache.resources["payroll-app"]) require.Len(t, grants, 1) @@ -100,7 +100,7 @@ func TestGrants_ExternalGrant_MatchID(t *testing.T) { cache, err := newSyncCache(ctx, data) require.NoError(t, err) - b := &resourceBuilder{cache: cache, resourceType: cache.resourceTypes["app"]} + b := testBuilder(cache, cache.resourceTypes["app"]) grants := allGrants(t, b, cache.resources["payroll-app"]) require.Len(t, grants, 1) @@ -125,7 +125,7 @@ func TestGrants_ExternalGrant_Expansion(t *testing.T) { cache, err := newSyncCache(ctx, data) require.NoError(t, err) - b := &resourceBuilder{cache: cache, resourceType: cache.resourceTypes["app"]} + b := testBuilder(cache, cache.resourceTypes["app"]) grants := allGrants(t, b, cache.resources["payroll-app"]) require.Len(t, grants, 1) @@ -156,7 +156,7 @@ func TestGrants_ExternalGrant_Expansion_DefaultDepthIsFull(t *testing.T) { cache, err := newSyncCache(ctx, data) require.NoError(t, err) - b := &resourceBuilder{cache: cache, resourceType: cache.resourceTypes["app"]} + b := testBuilder(cache, cache.resourceTypes["app"]) grants := allGrants(t, b, cache.resources["payroll-app"]) require.Len(t, grants, 1) @@ -190,7 +190,7 @@ func TestGrants_ExternalGrant_Expansion_SkipsInvalidCombos(t *testing.T) { cache, err := newSyncCache(ctx, data) require.NoError(t, err) - b := &resourceBuilder{cache: cache, resourceType: cache.resourceTypes["app"]} + b := testBuilder(cache, cache.resourceTypes["app"]) grants := allGrants(t, b, cache.resources["payroll-app"]) require.Empty(t, grants, "invalid expansion combinations must be skipped") } @@ -232,7 +232,7 @@ func TestGrants_ExternalGrant_SkipsInvalidRows(t *testing.T) { cache, err := newSyncCache(ctx, data) require.NoError(t, err) - b := &resourceBuilder{cache: cache, resourceType: cache.resourceTypes["app"]} + b := testBuilder(cache, cache.resourceTypes["app"]) grants := allGrants(t, b, cache.resources["payroll-app"]) require.Empty(t, grants, "invalid external grant rows must be skipped, not emitted") } diff --git a/pkg/connector/hot_reload_test.go b/pkg/connector/hot_reload_test.go new file mode 100644 index 00000000..d878ad68 --- /dev/null +++ b/pkg/connector/hot_reload_test.go @@ -0,0 +1,151 @@ +package connector + +import ( + "context" + "os" + "path/filepath" + "testing" + + v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" + "github.com/conductorone/baton-sdk/pkg/pagination" + rs "github.com/conductorone/baton-sdk/pkg/types/resource" + "github.com/stretchr/testify/require" +) + +// Hot-load contract tests (see cacheHolder in connector.go). +// +// The SDK constructs the connector and calls ResourceSyncers() exactly once +// per process, then re-runs Validate() at the start of every sync. In a +// long-running service, data-section changes to the input file MUST be +// served on the next sync without a restart. These tests mirror that +// lifecycle: one ResourceSyncers() call, then file edits followed by +// Validate(), asserting through the ORIGINAL builders. + +const hotReloadCSVBase = `record_type,id,display_name,email,status,type,profile.department,profile.title,resource_type,trait,description,resource_id,entitlement_slug,principal_id +user,alex.taylor,Alex Taylor,alex.taylor@example.com,enabled,human,Engineering,Software Engineer,,,,,, +user,sam.johnson,Sam Johnson,sam.johnson@example.com,enabled,human,Marketing,Marketing Manager,,,,,, +resource,engineering,Engineering,,,,,,team,group,Engineering team,,, +entitlement,,Member,,,,,,,,,engineering,member, +direct_user_grant,,,,,,,,,,,engineering,member,alex.taylor +` + +// hotReloadCSVUpdated adds one user and one grant to the base data. +const hotReloadCSVUpdated = hotReloadCSVBase + + `user,jordan.lee,Jordan Lee,jordan.lee@example.com,enabled,human,Engineering,Engineering Manager,,,,,, +direct_user_grant,,,,,,,,,,,engineering,member,jordan.lee +` + +// hotReloadCSVInvalid duplicates an existing user id, which fails +// ValidateUniqueIDs. +const hotReloadCSVInvalid = hotReloadCSVBase + + `user,alex.taylor,Alex Duplicate,alex.duplicate@example.com,enabled,human,,,,,,,, +` + +// startConnector writes the CSV, then walks the once-per-process SDK startup +// sequence (Validate, then ResourceSyncers) and returns the connector plus +// the long-lived builders keyed by resource type id. +func startConnector(t *testing.T, path, csv string) (*FileConnector, map[string]*resourceBuilder) { + t.Helper() + ctx := context.Background() + require.NoError(t, os.WriteFile(path, []byte(csv), 0o600)) + + fc := &FileConnector{inputFilePath: path} + _, err := fc.Validate(ctx) + require.NoError(t, err) + + builders := make(map[string]*resourceBuilder) + for _, s := range fc.ResourceSyncers(ctx) { + b, ok := s.(*resourceBuilder) + require.True(t, ok) + builders[b.ResourceType(ctx).GetId()] = b + } + require.Contains(t, builders, "user") + require.Contains(t, builders, "team") + return fc, builders +} + +func findResource(t *testing.T, b *resourceBuilder, id string) *v2.Resource { + t.Helper() + for _, r := range allResources(t, b, nil) { + if r.GetId().GetResource() == id { + return r + } + } + t.Fatalf("resource %q not found", id) + return nil +} + +func TestHotReload_DataChangesPickedUpBySync(t *testing.T) { + ctx := context.Background() + path := filepath.Join(t.TempDir(), "input.csv") + fc, builders := startConnector(t, path, hotReloadCSVBase) + + require.Len(t, allResources(t, builders["user"], nil), 2) + engineering := findResource(t, builders["team"], "engineering") + require.Len(t, allGrants(t, builders["team"], engineering), 1) + + // The customer edits the file while the service keeps running. + require.NoError(t, os.WriteFile(path, []byte(hotReloadCSVUpdated), 0o600)) + + // Validate() is the only per-sync hook the SDK gives this connector, so + // it alone must refresh the data the original builders serve. + _, err := fc.Validate(ctx) + require.NoError(t, err) + + require.Len(t, allResources(t, builders["user"], nil), 3) + require.Len(t, allGrants(t, builders["team"], engineering), 2) +} + +func TestHotReload_SwapMidListingRestartsPagination(t *testing.T) { + ctx := context.Background() + path := filepath.Join(t.TempDir(), "input.csv") + fc, builders := startConnector(t, path, hotReloadCSVBase) + + // Fetch the first page (1 of 2 users) so a token is in flight. + firstPage, results, err := builders["user"].List(ctx, nil, + rs.SyncOpAttrs{PageToken: pagination.Token{Size: 1}}) + require.NoError(t, err) + require.Len(t, firstPage, 1) + require.NotEmpty(t, results.NextPageToken) + + // The file changes and a Validate (sync start or health probe) swaps the + // cache while the listing is in flight. + require.NoError(t, os.WriteFile(path, []byte(hotReloadCSVUpdated), 0o600)) + _, err = fc.Validate(ctx) + require.NoError(t, err) + + // Resuming with the stale token must restart the listing against the new + // generation — every current user is returned exactly once, rather than + // offsets silently skipping or duplicating rows across generations. + seen := map[string]int{} + token := results.NextPageToken + for { + page, res, err := builders["user"].List(ctx, nil, + rs.SyncOpAttrs{PageToken: pagination.Token{Token: token, Size: 1}}) + require.NoError(t, err) + for _, r := range page { + seen[r.GetId().GetResource()]++ + } + if token = res.NextPageToken; token == "" { + break + } + } + require.Equal(t, map[string]int{"alex.taylor": 1, "sam.johnson": 1, "jordan.lee": 1}, seen) +} + +func TestHotReload_InvalidFileKeepsLastGoodCache(t *testing.T) { + ctx := context.Background() + path := filepath.Join(t.TempDir(), "input.csv") + fc, builders := startConnector(t, path, hotReloadCSVBase) + + require.Len(t, allResources(t, builders["user"], nil), 2) + + // The file is replaced with one that fails validation: the sync must + // fail loudly while the last-known-good data keeps being served. + require.NoError(t, os.WriteFile(path, []byte(hotReloadCSVInvalid), 0o600)) + + _, err := fc.Validate(ctx) + require.Error(t, err) + + require.Len(t, allResources(t, builders["user"], nil), 2) +} diff --git a/pkg/connector/resources.go b/pkg/connector/resources.go index d3f564a0..60d1f99f 100644 --- a/pkg/connector/resources.go +++ b/pkg/connector/resources.go @@ -24,21 +24,53 @@ func clampPageSize(requested int) int { return requested } -func paginate[T any](items []T, tokenStr string, tokenSize int) ([]T, string, error) { +// parseOffset validates a numeric page-token offset. +func parseOffset(tokenStr, offsetStr string) (int, error) { + parsed, err := strconv.Atoi(offsetStr) + if err != nil { + return 0, fmt.Errorf("baton-file: invalid page token %q: %w", tokenStr, err) + } + if parsed <= 0 { + return 0, fmt.Errorf("baton-file: invalid page token %q: must be a positive integer", tokenStr) + } + return parsed, nil +} + +func paginate[T any](items []T, gen string, tokenStr string, tokenSize int) ([]T, string, error) { // SDK sends Size=0 during sync_full; clamp converts it to pageSize so we never return an empty page. size := clampPageSize(tokenSize) - // Empty token means first page; non-empty token is the numeric offset where this page starts. + // Empty token means first page. Tokens minted by this version are + // ":" where gen fingerprints the cache the offset indexes + // into; bare numeric tokens are the legacy (pre-generation) format. offset := 0 if tokenStr != "" { - parsed, err := strconv.Atoi(tokenStr) - if err != nil { - return nil, "", fmt.Errorf("baton-file: invalid page token %q: %w", tokenStr, err) + tokenGen, offsetStr, found := strings.Cut(tokenStr, ":") + switch { + case !found: + // Legacy numeric token: honor it as a plain offset so an + // in-flight sync survives a binary upgrade. + parsed, err := parseOffset(tokenStr, tokenStr) + if err != nil { + return nil, "", err + } + offset = parsed + case tokenGen != gen: + // The token was minted against a different cache generation: the + // file changed and the cache was swapped while this listing was + // in flight (mid-sync health-check revalidation, or a sync + // resumed after a restart). Offsets are meaningless across + // generations — resuming would silently skip or duplicate items. + // Restart the listing instead: store writes are idempotent + // upserts, so re-emitting earlier pages is safe. + offset = 0 + default: + parsed, err := parseOffset(tokenStr, offsetStr) + if err != nil { + return nil, "", err + } + offset = parsed } - if parsed <= 0 { - return nil, "", fmt.Errorf("baton-file: invalid page token %q: must be a positive integer", tokenStr) - } - offset = parsed } // Guard against a stale checkpoint: if the file was replaced with fewer @@ -54,16 +86,26 @@ func paginate[T any](items []T, tokenStr string, tokenSize int) ([]T, string, er end = len(items) } - // Empty next token signals the SDK that there are no more pages. + // Empty next token signals the SDK that there are no more pages. An empty + // gen (caches built directly in tests) mints legacy numeric tokens. var next string if end < len(items) { - next = strconv.Itoa(end) + if gen == "" { + next = strconv.Itoa(end) + } else { + next = gen + ":" + strconv.Itoa(end) + } } return items[offset:end], next, nil } type resourceBuilder struct { - cache *syncCache + // cache is the shared live holder, NOT a snapshot. Builders live for the + // whole process while Validate() republishes the cache each sync. + // IMPORTANT: do not replace this with a *syncCache captured at + // construction — that freezes file data until the service restarts + // (hot-load regression; see cacheHolder in connector.go). + cache *cacheHolder resourceType *v2.ResourceType } @@ -73,8 +115,12 @@ func (b *resourceBuilder) ResourceType(ctx context.Context) *v2.ResourceType { func (b *resourceBuilder) List(_ context.Context, parentResourceID *v2.ResourceId, opts rs.SyncOpAttrs) ([]*v2.Resource, *rs.SyncOpResults, error) { - resources := b.cache.listIndex[listKey(b.resourceType.GetId(), parentResourceID)] - page, next, err := paginate(resources, opts.PageToken.Token, opts.PageToken.Size) + cache := b.cache.load() + if cache == nil { + return nil, nil, fmt.Errorf("baton-file: sync cache not initialized") + } + resources := cache.listIndex[listKey(b.resourceType.GetId(), parentResourceID)] + page, next, err := paginate(resources, cache.gen, opts.PageToken.Token, opts.PageToken.Size) if err != nil { return nil, nil, err } @@ -83,8 +129,12 @@ func (b *resourceBuilder) List(_ context.Context, parentResourceID *v2.ResourceI func (b *resourceBuilder) Entitlements(_ context.Context, resource *v2.Resource, opts rs.SyncOpAttrs) ([]*v2.Entitlement, *rs.SyncOpResults, error) { - ents := b.cache.entIndex[resource.GetId().GetResource()] - page, next, err := paginate(ents, opts.PageToken.Token, opts.PageToken.Size) + cache := b.cache.load() + if cache == nil { + return nil, nil, fmt.Errorf("baton-file: sync cache not initialized") + } + ents := cache.entIndex[resource.GetId().GetResource()] + page, next, err := paginate(ents, cache.gen, opts.PageToken.Token, opts.PageToken.Size) if err != nil { return nil, nil, err } @@ -93,8 +143,12 @@ func (b *resourceBuilder) Entitlements(_ context.Context, resource *v2.Resource, func (b *resourceBuilder) Grants(_ context.Context, resource *v2.Resource, opts rs.SyncOpAttrs) ([]*v2.Grant, *rs.SyncOpResults, error) { - grants := b.cache.grantsIndex[resource.GetId().GetResource()] - page, next, err := paginate(grants, opts.PageToken.Token, opts.PageToken.Size) + cache := b.cache.load() + if cache == nil { + return nil, nil, fmt.Errorf("baton-file: sync cache not initialized") + } + grants := cache.grantsIndex[resource.GetId().GetResource()] + page, next, err := paginate(grants, cache.gen, opts.PageToken.Token, opts.PageToken.Size) if err != nil { return nil, nil, err } @@ -110,7 +164,7 @@ func buildUserResource(ctx context.Context, userData client.UserData, opts = append(opts, rs.WithEmail(userData.Email, true)) } if len(userData.Profile) > 0 { - opts = append(opts, rs.WithUserProfile(userData.Profile)) + opts = append(opts, rs.WithUserProfile(userData.Profile)) //nolint:staticcheck // kept: sets trait+resource field; replacement drops the trait field, changing sync output } userStatus := v2.UserTrait_Status_STATUS_ENABLED @@ -175,9 +229,9 @@ func buildUserResource(ctx context.Context, userData client.UserData, } if userData.StatusDetails != "" { - opts = append(opts, rs.WithDetailedStatus(userStatus, userData.StatusDetails)) + opts = append(opts, rs.WithDetailedStatus(userStatus, userData.StatusDetails)) //nolint:staticcheck // kept: sets trait+resource field; replacement drops the trait field, changing sync output } else { - opts = append(opts, rs.WithStatus(userStatus)) + opts = append(opts, rs.WithStatus(userStatus)) //nolint:staticcheck // kept: sets trait+resource field; replacement drops the trait field, changing sync output } return rs.NewUserResource(userData.DisplayName, resourceType, userData.ID, opts) @@ -212,7 +266,7 @@ func buildResource(ctx context.Context, data client.ResourceData, func buildGroupTraitOptions(data client.ResourceData) []rs.GroupTraitOption { var opts []rs.GroupTraitOption if len(data.Profile) > 0 { - opts = append(opts, rs.WithGroupProfile(data.Profile)) + opts = append(opts, rs.WithGroupProfile(data.Profile)) //nolint:staticcheck // kept: sets trait+resource field; replacement drops the trait field, changing sync output } return opts } @@ -220,7 +274,7 @@ func buildGroupTraitOptions(data client.ResourceData) []rs.GroupTraitOption { func buildRoleTraitOptions(data client.ResourceData) []rs.RoleTraitOption { var opts []rs.RoleTraitOption if len(data.Profile) > 0 { - opts = append(opts, rs.WithRoleProfile(data.Profile)) + opts = append(opts, rs.WithRoleProfile(data.Profile)) //nolint:staticcheck // kept: sets trait+resource field; replacement drops the trait field, changing sync output } return opts } @@ -228,7 +282,7 @@ func buildRoleTraitOptions(data client.ResourceData) []rs.RoleTraitOption { func buildAppTraitOptions(data client.ResourceData) []rs.AppTraitOption { var opts []rs.AppTraitOption if len(data.Profile) > 0 { - opts = append(opts, rs.WithAppProfile(data.Profile)) + opts = append(opts, rs.WithAppProfile(data.Profile)) //nolint:staticcheck // kept: sets trait+resource field; replacement drops the trait field, changing sync output } return opts } @@ -243,7 +297,7 @@ func buildSecretTraitOptions(ctx context.Context, data client.ResourceData) []rs l.Warn("baton-file: failed to parse created_at for secret, skipping", zap.String("resource_id", data.ID), zap.Error(err)) } else { - opts = append(opts, rs.WithSecretCreatedAt(*t)) + opts = append(opts, rs.WithSecretCreatedAt(*t)) //nolint:staticcheck // kept: sets trait+resource field; replacement drops the trait field, changing sync output } } diff --git a/pkg/connector/resources_test.go b/pkg/connector/resources_test.go index c710d8cb..03dfa3ed 100644 --- a/pkg/connector/resources_test.go +++ b/pkg/connector/resources_test.go @@ -12,6 +12,14 @@ import ( "github.com/stretchr/testify/require" ) +// testBuilder wraps a pre-built cache in a holder, mirroring what +// ResourceSyncers does with the live connector cache. +func testBuilder(cache *syncCache, rt *v2.ResourceType) *resourceBuilder { + h := &cacheHolder{} + h.store(cache) + return &resourceBuilder{cache: h, resourceType: rt} +} + // allGrants pages through Grants() until exhausted and returns every grant. func allGrants(t *testing.T, b *resourceBuilder, res *v2.Resource) []*v2.Grant { t.Helper() @@ -97,7 +105,7 @@ func TestPaginate_Exhaustion(t *testing.T) { token := "" pages := 0 for { - page, next, err := paginate(items, token, 0) + page, next, err := paginate(items, "", token, 0) require.NoError(t, err) require.LessOrEqual(t, len(page), pageSize) pages++ @@ -117,7 +125,7 @@ func TestPaginate_Exhaustion(t *testing.T) { func TestPaginate_StaleOffset(t *testing.T) { items := make([]int, 50) - page, next, err := paginate(items, "200", 0) + page, next, err := paginate(items, "", "200", 0) require.NoError(t, err) require.Empty(t, page) require.Empty(t, next) @@ -125,12 +133,48 @@ func TestPaginate_StaleOffset(t *testing.T) { func TestPaginate_InvalidToken(t *testing.T) { items := []int{1, 2, 3} - for _, tok := range []string{"not-a-number", "abc", "-1", "0"} { - _, _, err := paginate(items, tok, 0) + for _, tok := range []string{"not-a-number", "abc", "-1", "0", "gen1:abc", "gen1:-1", "gen1:0"} { + _, _, err := paginate(items, "gen1", tok, 0) require.Error(t, err, "token %q should return error", tok) } } +func TestPaginate_GenerationStampedTokens(t *testing.T) { + items := []int{10, 20, 30} + + // Tokens minted under a gen carry it; same-gen resume honors the offset. + page, next, err := paginate(items, "gen1", "", 2) + require.NoError(t, err) + require.Equal(t, []int{10, 20}, page) + require.Equal(t, "gen1:2", next) + + page, next, err = paginate(items, "gen1", next, 2) + require.NoError(t, err) + require.Equal(t, []int{30}, page) + require.Empty(t, next) +} + +func TestPaginate_GenerationMismatchRestartsListing(t *testing.T) { + // A token minted against a different cache generation must restart the + // listing at the beginning instead of replaying its offset — offsets are + // meaningless across generations (hot-load swapped the cache mid-listing). + items := []int{10, 20, 30} + page, next, err := paginate(items, "gen2", "gen1:2", 2) + require.NoError(t, err) + require.Equal(t, []int{10, 20}, page) + require.Equal(t, "gen2:2", next) +} + +func TestPaginate_LegacyNumericTokenHonored(t *testing.T) { + // Bare numeric tokens (minted by a pre-generation binary) resume as plain + // offsets so an in-flight sync survives a binary upgrade. + items := []int{10, 20, 30} + page, next, err := paginate(items, "gen1", "2", 2) + require.NoError(t, err) + require.Equal(t, []int{30}, page) + require.Empty(t, next) +} + // Grants() behavior tests. func TestGrants_InheritanceMapping_ExpandableAnnotation(t *testing.T) { @@ -157,7 +201,7 @@ func TestGrants_InheritanceMapping_ExpandableAnnotation(t *testing.T) { cache, err := newSyncCache(ctx, data) require.NoError(t, err) - b := &resourceBuilder{cache: cache, resourceType: cache.resourceTypes["role"]} + b := testBuilder(cache, cache.resourceTypes["role"]) grants := allGrants(t, b, cache.resources["role-x"]) require.Len(t, grants, 1) @@ -197,7 +241,7 @@ func TestGrants_PaginatesCorrectly(t *testing.T) { cache, err := newSyncCache(ctx, data) require.NoError(t, err) - b := &resourceBuilder{cache: cache, resourceType: cache.resourceTypes["group"]} + b := testBuilder(cache, cache.resourceTypes["group"]) grants := allGrants(t, b, cache.resources["res1"]) require.Len(t, grants, total) @@ -221,7 +265,7 @@ func TestGrants_InvalidPageToken(t *testing.T) { cache, err := newSyncCache(ctx, data) require.NoError(t, err) - b := &resourceBuilder{cache: cache, resourceType: cache.resourceTypes["group"]} + b := testBuilder(cache, cache.resourceTypes["group"]) res := cache.resources["res1"] for _, tok := range []string{"not-a-number", "-1", "0"} { @@ -251,7 +295,7 @@ func TestEntitlements_PaginatesCorrectly(t *testing.T) { cache, err := newSyncCache(ctx, data) require.NoError(t, err) - b := &resourceBuilder{cache: cache, resourceType: cache.resourceTypes["role"]} + b := testBuilder(cache, cache.resourceTypes["role"]) ents := allEntitlements(t, b, cache.resources["res1"]) require.Len(t, ents, total) @@ -278,7 +322,7 @@ func TestEntitlements_OnlyForRequestedResource(t *testing.T) { cache, err := newSyncCache(ctx, data) require.NoError(t, err) - b := &resourceBuilder{cache: cache, resourceType: cache.resourceTypes["group"]} + b := testBuilder(cache, cache.resourceTypes["group"]) ents := allEntitlements(t, b, cache.resources["eng"]) require.Len(t, ents, 1, "must only return entitlements for the requested resource") @@ -303,7 +347,7 @@ func TestList_PaginatesCorrectly(t *testing.T) { cache, err := newSyncCache(ctx, data) require.NoError(t, err) - b := &resourceBuilder{cache: cache, resourceType: cache.resourceTypes["group"]} + b := testBuilder(cache, cache.resourceTypes["group"]) got := allResources(t, b, nil) require.Len(t, got, total) @@ -327,7 +371,7 @@ func TestList_ReturnsCorrectResourceType(t *testing.T) { cache, err := newSyncCache(ctx, data) require.NoError(t, err) - b := &resourceBuilder{cache: cache, resourceType: cache.resourceTypes["group"]} + b := testBuilder(cache, cache.resourceTypes["group"]) resources := allResources(t, b, nil) require.Len(t, resources, 1) @@ -348,7 +392,7 @@ func TestList_FiltersByParent(t *testing.T) { cache, err := newSyncCache(ctx, data) require.NoError(t, err) - b := &resourceBuilder{cache: cache, resourceType: cache.resourceTypes["team"]} + b := testBuilder(cache, cache.resourceTypes["team"]) topLevel := allResources(t, b, nil) require.Len(t, topLevel, 1) From faf2f70244bc6e36fe50ced80966d87e93f445b9 Mon Sep 17 00:00:00 2001 From: Matthew Scobell Date: Thu, 13 Aug 2026 18:37:25 -0400 Subject: [PATCH 02/11] docs: state that resource types are intentionally frozen at registration Hot-load covers the file's data sections only; the set of resource types and their traits is schema, fixed for the process lifetime because the SDK registers syncers by type exactly once. Document that intent at the builder-registration site and widen the README/contract wording from "a new resource type" to any resource-type or trait change. Co-Authored-By: Claude Fable 5 --- README.md | 2 +- pkg/connector/connector.go | 11 ++++++++++- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 763d523a..a66f8bd7 100644 --- a/README.md +++ b/README.md @@ -41,7 +41,7 @@ baton-file -i data.xlsx ## Hot-Load (File Change Detection) -When baton-file runs as a long-lived service, edits to the input file are picked up automatically at the start of the next sync cycle — no restart required. This applies to the **data** sections: users, resources, entitlements, grants, and inheritance mappings. **Schema changes** — introducing a brand-new resource type — still require a service restart, because the SDK registers resource types once at process startup. If the edited file fails validation, the sync fails with the validation error and the connector keeps serving the last successfully loaded data. If the file changes while a sync is in flight (for example via health-check revalidation), affected listings restart safely against the new contents instead of resuming into stale page offsets. +When baton-file runs as a long-lived service, edits to the input file are picked up automatically at the start of the next sync cycle — no restart required. This applies to the **data** sections: users, resources, entitlements, grants, and inheritance mappings. **Schema changes** — introducing a brand-new resource type, or changing an existing type's trait — still require a service restart, because the SDK registers resource types and their traits once at process startup. If the edited file fails validation, the sync fails with the validation error and the connector keeps serving the last successfully loaded data. If the file changes while a sync is in flight (for example via health-check revalidation), affected listings restart safely against the new contents instead of resuming into stale page offsets. **IMPORTANT (maintainers and AI agents):** this behavior is a deliberate contract, not an accident of implementation. The SDK calls `ResourceSyncers()` exactly once per process, so `Validate()` — which runs at the start of every sync — is the only per-sync hook and is where the file is re-read and the shared cache is republished (see `cacheHolder` in `pkg/connector/connector.go`). Do not move the file load to construction time, capture cache snapshots in builders, or assume the SDK refreshes data between syncs (it does not — that assumption caused a regression once already). `TestHotReload_DataChangesPickedUpBySync` in `pkg/connector/hot_reload_test.go` enforces this contract; if it fails after your change, the change is wrong, not the test. diff --git a/pkg/connector/connector.go b/pkg/connector/connector.go index be83d5e9..656677dc 100644 --- a/pkg/connector/connector.go +++ b/pkg/connector/connector.go @@ -40,7 +40,8 @@ type FileConnector struct { // only per-sync hook this connector gets is Validate(), so Validate() is // where the input file is re-read and the fresh syncCache is published here. // Data-section changes to the file MUST be picked up on the next sync without -// a restart; only schema changes (a new resource type) require one. Do NOT +// a restart; only schema changes (a new resource type, or a trait change to +// an existing type) require one. Do NOT // move the file load back to construction time, capture a *syncCache snapshot // in a builder, or assume the SDK refreshes anything between syncs (it does // not — that assumption caused the original hot-load regression in PR #40). @@ -244,6 +245,14 @@ func (fc *FileConnector) ResourceSyncers(ctx context.Context) []connectorbuilder var syncers []connectorbuilder.ResourceSyncerV2 for _, rt := range cache.resourceTypes { + // resourceType is INTENTIONALLY frozen at registration time and never + // refreshed from later cache generations. Hot-load covers the file's + // data sections only; schema — the set of resource types and their + // traits — is fixed for the process lifetime because the SDK registers + // syncers by type exactly once. Editing an existing type's trait in + // the file therefore requires a restart, same as adding a new type; + // resolving rt from the live cache here would not change that, since + // the SDK-side registration would still hold the startup-time type. syncers = append(syncers, &resourceBuilder{cache: &fc.cache, resourceType: rt}) } From 6cf147468fd6b6cc2b3ef1e7befa9c5d9a06d09d Mon Sep 17 00:00:00 2001 From: Matthew Scobell Date: Thu, 13 Aug 2026 18:40:29 -0400 Subject: [PATCH 03/11] docs: explain the why behind intentional design decisions inline Add rationale comments at every site where this connector deviates from standard patterns, for reviewers and future maintainers: the nil holder on capabilities-only builders, the deprecated trait options kept to preserve trait-level sync output, the raw-bytes fingerprint and its benign double-read, the Debug/Warn log-level choices, and why hot-load tests use real files instead of in-memory data. Co-Authored-By: Claude Fable 5 --- pkg/connector/connector.go | 13 +++++++++++-- pkg/connector/hot_reload_test.go | 4 ++++ pkg/connector/resources.go | 13 +++++++++++++ 3 files changed, 28 insertions(+), 2 deletions(-) diff --git a/pkg/connector/connector.go b/pkg/connector/connector.go index 656677dc..4366cc35 100644 --- a/pkg/connector/connector.go +++ b/pkg/connector/connector.go @@ -169,6 +169,8 @@ func (fc *FileConnector) Validate(ctx context.Context) (annotations.Annotations, } fc.cache.store(cache) + // Debug, not Info: this fires on every sync AND every health-check + // probe, so Info would flood default logs on probed deployments. ctxzap.Extract(ctx).Debug("baton-file: refreshed data from input file", zap.Int("resource_types", len(cache.resourceTypes)), zap.Int("resources", len(cache.resources))) @@ -179,8 +181,12 @@ func (fc *FileConnector) Validate(ctx context.Context) (annotations.Annotations, // loadValidatedCache reads the input file, runs all cross-record validations, // and builds the derived sync cache from it. func loadValidatedCache(ctx context.Context, inputFilePath string) (*syncCache, error) { - // Fingerprint the raw bytes before parsing; a file edit racing between - // the two reads mislabels one build, which at worst restarts a listing. + // Fingerprint the raw bytes to stamp this build's generation into page + // tokens (see paginate). Hashing raw content is format-agnostic and + // exact, and the extra read is cheap next to parsing — the loaders take + // paths, not bytes, so reusing one read would mean restructuring them. A + // file edit racing between the two reads mislabels one build, which at + // worst restarts a listing. raw, err := os.ReadFile(inputFilePath) if err != nil { return nil, fmt.Errorf("baton-file: failed to read input file: %w", err) @@ -234,6 +240,9 @@ func (fc *FileConnector) ResourceSyncers(ctx context.Context) []connectorbuilder // fallback covers entry points that skip Validate(). This method's // signature cannot return an error, so log-and-return-nil is // intentional — the connector stays alive for the next sync cycle. + // Warn, not Error: Validate() returns the real error to the SDK on + // every sync, so this log is a secondary signal, and house rules + // classify input/config failures as Warn. var err error cache, err = loadValidatedCache(ctx, fc.inputFilePath) if err != nil { diff --git a/pkg/connector/hot_reload_test.go b/pkg/connector/hot_reload_test.go index d878ad68..51dcd94e 100644 --- a/pkg/connector/hot_reload_test.go +++ b/pkg/connector/hot_reload_test.go @@ -20,6 +20,10 @@ import ( // served on the next sync without a restart. These tests mirror that // lifecycle: one ResourceSyncers() call, then file edits followed by // Validate(), asserting through the ORIGINAL builders. +// +// Unlike the other connector tests, fixtures here are real files written to +// disk rather than in-memory LoadedData: hot-load IS the file → Validate → +// cache path, so bypassing the loaders would test nothing. const hotReloadCSVBase = `record_type,id,display_name,email,status,type,profile.department,profile.title,resource_type,trait,description,resource_id,entitlement_slug,principal_id user,alex.taylor,Alex Taylor,alex.taylor@example.com,enabled,human,Engineering,Software Engineer,,,,,, diff --git a/pkg/connector/resources.go b/pkg/connector/resources.go index 60d1f99f..3e62042e 100644 --- a/pkg/connector/resources.go +++ b/pkg/connector/resources.go @@ -105,6 +105,11 @@ type resourceBuilder struct { // IMPORTANT: do not replace this with a *syncCache captured at // construction — that freezes file data until the service restarts // (hot-load regression; see cacheHolder in connector.go). + // + // It is nil for builders made by StaticCapabilitiesConnector: the + // capabilities sub-command registers resource types without reading any + // file. No SDK path lists through those builders, but the sync methods + // below still guard the nil and return an error instead of panicking. cache *cacheHolder resourceType *v2.ResourceType } @@ -155,6 +160,14 @@ func (b *resourceBuilder) Grants(_ context.Context, resource *v2.Resource, return page, &rs.SyncOpResults{NextPageToken: next}, nil } +// The deprecated trait options used below (WithUserProfile, WithStatus, +// WithGroupProfile, WithSecretCreatedAt, ...) are kept INTENTIONALLY. The SDK +// is migrating profile/status/created_at from the trait protos to +// resource-level attributes: the deprecated options populate BOTH levels, +// while the WithResource* replacements populate only the resource level. +// A lint-driven swap would silently drop the trait-level fields from sync +// output — breaking anything downstream that still reads traits — so the +// migration needs its own coordinated change, not a mechanical rename. func buildUserResource(ctx context.Context, userData client.UserData, resourceType *v2.ResourceType) (*v2.Resource, error) { l := ctxzap.Extract(ctx) From 14889eb8a02c5b001a55d96d18b1b6a4acb4d427 Mon Sep 17 00:00:00 2001 From: Matthew Scobell Date: Thu, 13 Aug 2026 21:10:56 -0400 Subject: [PATCH 04/11] =?UTF-8?q?fix:=20address=20review=20suggestions=20?= =?UTF-8?q?=E2=80=94=20legacy-token=20restart,=20gen=20short-circuit?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Legacy bare-numeric page tokens now restart the listing when the cache is generation-stamped: their generation is unknowable and they only appear after a restart, exactly when the file may have changed. Generation-less caches (built directly in tests) still honor their own numeric mints — restarting those would loop forever. - loadValidatedCache short-circuits when the content fingerprint matches the published cache generation: skips the redundant parse and rebuild on health-check probes and keeps no-op Validate calls from swapping the pointer under an in-flight sync. - Document the cross-listing consistency limit in the cacheHolder contract: generation stamps make each listing consistent, not a whole sync; the short-circuit confines swaps to actual content changes. - Add the hot-load note to the customer-facing docs/connector.mdx. Co-Authored-By: Claude Fable 5 --- docs/connector.mdx | 1 + pkg/connector/connector.go | 45 ++++++++++++++++++++++++-------- pkg/connector/hot_reload_test.go | 21 +++++++++++++++ pkg/connector/resources.go | 17 +++++++++--- pkg/connector/resources_test.go | 16 +++++++++--- 5 files changed, 83 insertions(+), 17 deletions(-) diff --git a/docs/connector.mdx b/docs/connector.mdx index c66308b0..f0c99d81 100644 --- a/docs/connector.mdx +++ b/docs/connector.mdx @@ -34,6 +34,7 @@ Resource types are defined by the input file (via each row's `trait`), so the ex - **Read-only.** The connector cannot create, modify, revoke, or deprovision access in the source system — it only reads what the file describes. - **Full sync only.** Every sync reads the entire file; there is no incremental/targeted sync. - **File-defined data.** C1 only knows what the file contains. Resource IDs must be globally unique across users and resources. +- **File edits apply on the next sync.** When the connector runs as a long-lived service, changes to the file's data — users, resources, entitlements, grants, and inheritance mappings — are picked up automatically at the start of the next sync, with no restart needed. Adding a new resource type, or changing an existing type's trait, does require restarting the connector. If an edited file fails validation, the sync fails with the validation error and the connector keeps serving the last successfully loaded data. ## Gather File connector credentials diff --git a/pkg/connector/connector.go b/pkg/connector/connector.go index 4366cc35..4e9a5587 100644 --- a/pkg/connector/connector.go +++ b/pkg/connector/connector.go @@ -56,6 +56,17 @@ type FileConnector struct { // cache generation (syncCache.gen) so an in-flight listing restarts instead // of resuming into a different generation's offsets. During a swap both old // and new caches are briefly live (~2x peak memory for the dataset). +// +// Known limit: generation stamps make each individual listing consistent, +// not a whole sync. If the file genuinely changes mid-sync and a health +// probe swaps the cache, phases that already ran (e.g. List) reflect the old +// generation while later phases (e.g. Grants) reflect the new one — a +// one-sync inconsistency (such as a grant whose principal was never listed) +// that the next sync heals. The connector cannot pin a cache per sync: a +// probe Validate and a sync-start Validate arrive as the same RPC, and no +// sync-lifecycle hook exists. The fingerprint short-circuit in +// loadValidatedCache confines this to actual content changes — unchanged +// files never swap. type cacheHolder struct { p atomic.Pointer[syncCache] } @@ -163,24 +174,31 @@ func (fc *FileConnector) Validate(ctx context.Context) (annotations.Annotations, // runs at the start of every sync, so this is the hot-load refresh point // (see cacheHolder). On error we return without storing, so the // previously published cache keeps serving last-known-good data. - cache, err := loadValidatedCache(ctx, fc.inputFilePath) + previous := fc.cache.load() + cache, err := loadValidatedCache(ctx, fc.inputFilePath, previous) if err != nil { return nil, err } - fc.cache.store(cache) + if cache != previous { + fc.cache.store(cache) - // Debug, not Info: this fires on every sync AND every health-check - // probe, so Info would flood default logs on probed deployments. - ctxzap.Extract(ctx).Debug("baton-file: refreshed data from input file", - zap.Int("resource_types", len(cache.resourceTypes)), - zap.Int("resources", len(cache.resources))) + // Debug, not Info: this fires on every sync AND every health-check + // probe, so Info would flood default logs on probed deployments. + ctxzap.Extract(ctx).Debug("baton-file: refreshed data from input file", + zap.Int("resource_types", len(cache.resourceTypes)), + zap.Int("resources", len(cache.resources))) + } return nil, nil } // loadValidatedCache reads the input file, runs all cross-record validations, -// and builds the derived sync cache from it. -func loadValidatedCache(ctx context.Context, inputFilePath string) (*syncCache, error) { +// and builds the derived sync cache from it. When the file's content +// fingerprint matches the current cache's generation, current is returned +// unchanged: the rebuild would be identical, and skipping it both avoids +// redundant work on health-check probes and keeps no-op Validate calls from +// swapping the pointer under an in-flight sync. +func loadValidatedCache(ctx context.Context, inputFilePath string, current *syncCache) (*syncCache, error) { // Fingerprint the raw bytes to stamp this build's generation into page // tokens (see paginate). Hashing raw content is format-agnostic and // exact, and the extra read is cheap next to parsing — the loaders take @@ -192,6 +210,11 @@ func loadValidatedCache(ctx context.Context, inputFilePath string) (*syncCache, return nil, fmt.Errorf("baton-file: failed to read input file: %w", err) } sum := sha256.Sum256(raw) + gen := hex.EncodeToString(sum[:8]) + + if current != nil && current.gen == gen { + return current, nil + } data, err := client.LoadFileData(inputFilePath) if err != nil { @@ -222,7 +245,7 @@ func loadValidatedCache(ctx context.Context, inputFilePath string) (*syncCache, if err != nil { return nil, err } - cache.gen = hex.EncodeToString(sum[:8]) + cache.gen = gen return cache, nil } @@ -244,7 +267,7 @@ func (fc *FileConnector) ResourceSyncers(ctx context.Context) []connectorbuilder // every sync, so this log is a secondary signal, and house rules // classify input/config failures as Warn. var err error - cache, err = loadValidatedCache(ctx, fc.inputFilePath) + cache, err = loadValidatedCache(ctx, fc.inputFilePath, nil) if err != nil { l.Warn("baton-file: failed to load input file", zap.Error(err)) return nil diff --git a/pkg/connector/hot_reload_test.go b/pkg/connector/hot_reload_test.go index 51dcd94e..ad733fdc 100644 --- a/pkg/connector/hot_reload_test.go +++ b/pkg/connector/hot_reload_test.go @@ -137,6 +137,27 @@ func TestHotReload_SwapMidListingRestartsPagination(t *testing.T) { require.Equal(t, map[string]int{"alex.taylor": 1, "sam.johnson": 1, "jordan.lee": 1}, seen) } +func TestHotReload_UnchangedFileSkipsRebuildAndSwap(t *testing.T) { + ctx := context.Background() + path := filepath.Join(t.TempDir(), "input.csv") + fc, _ := startConnector(t, path, hotReloadCSVBase) + + // A Validate against unchanged content (every health probe, and every + // sync where nobody edited the file) must return the already-published + // cache: no rebuild, and critically no pointer swap under an in-flight + // sync. Pointer identity is the contract. + before := fc.cache.load() + _, err := fc.Validate(ctx) + require.NoError(t, err) + require.Same(t, before, fc.cache.load()) + + // A real change still swaps. + require.NoError(t, os.WriteFile(path, []byte(hotReloadCSVUpdated), 0o600)) + _, err = fc.Validate(ctx) + require.NoError(t, err) + require.NotSame(t, before, fc.cache.load()) +} + func TestHotReload_InvalidFileKeepsLastGoodCache(t *testing.T) { ctx := context.Background() path := filepath.Join(t.TempDir(), "input.csv") diff --git a/pkg/connector/resources.go b/pkg/connector/resources.go index 3e62042e..7d1655a1 100644 --- a/pkg/connector/resources.go +++ b/pkg/connector/resources.go @@ -48,13 +48,24 @@ func paginate[T any](items []T, gen string, tokenStr string, tokenSize int) ([]T tokenGen, offsetStr, found := strings.Cut(tokenStr, ":") switch { case !found: - // Legacy numeric token: honor it as a plain offset so an - // in-flight sync survives a binary upgrade. + // Bare numeric token. If this cache has a generation (every + // production cache does — loadValidatedCache always stamps one), + // the token is legacy, minted by a pre-generation binary. Its + // generation is unknowable and it can only arrive after a + // process restart — exactly when the file may have changed while + // the service was down — so restart the listing like a + // generation mismatch rather than replay an offset into a slice + // it was never minted against. A generation-less cache (tests + // build them directly) mints bare tokens itself, so there the + // offset is trusted; restarting would loop forever. Malformed + // tokens are rejected either way. parsed, err := parseOffset(tokenStr, tokenStr) if err != nil { return nil, "", err } - offset = parsed + if gen == "" { + offset = parsed + } case tokenGen != gen: // The token was minted against a different cache generation: the // file changed and the cache was swapped while this listing was diff --git a/pkg/connector/resources_test.go b/pkg/connector/resources_test.go index 03dfa3ed..d3045c86 100644 --- a/pkg/connector/resources_test.go +++ b/pkg/connector/resources_test.go @@ -165,12 +165,22 @@ func TestPaginate_GenerationMismatchRestartsListing(t *testing.T) { require.Equal(t, "gen2:2", next) } -func TestPaginate_LegacyNumericTokenHonored(t *testing.T) { - // Bare numeric tokens (minted by a pre-generation binary) resume as plain - // offsets so an in-flight sync survives a binary upgrade. +func TestPaginate_LegacyNumericTokenRestartsListing(t *testing.T) { + // Against a generation-stamped cache, bare numeric tokens (minted by a + // pre-generation binary) restart the listing: their generation is + // unknowable and they only appear after a restart, when the file may + // have changed. Malformed ones still error (see TestPaginate_InvalidToken). items := []int{10, 20, 30} page, next, err := paginate(items, "gen1", "2", 2) require.NoError(t, err) + require.Equal(t, []int{10, 20}, page) + require.Equal(t, "gen1:2", next) + + // Against a generation-less cache (tests build these directly), bare + // numeric tokens are the cache's own mint format and must be honored — + // restarting would loop forever. + page, next, err = paginate(items, "", "2", 2) + require.NoError(t, err) require.Equal(t, []int{30}, page) require.Empty(t, next) } From 8daf34c2520ad0272cfbb321a7c2dd20548a7ff1 Mon Sep 17 00:00:00 2001 From: Matthew Scobell Date: Thu, 13 Aug 2026 21:25:58 -0400 Subject: [PATCH 05/11] fix: serialize cache refresh; correct lifecycle docs and test fidelity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Serialize the read -> build -> publish sequence behind a mutex (extracted as FileConnector.refresh): the atomic pointer prevented data races but not lost updates — two concurrent Validate calls racing across two file edits could publish out of order. A CAS loop cannot fix this because generations are unordered content hashes; serializing makes publish order follow read order and prevents concurrent full cache builds. - Correct the ResourceSyncers comment, which had the lifecycle backwards: the SDK calls it at construction BEFORE any Validate, so the construction-time load is the first load, not a fallback. Document the honest consequence: a file that is invalid at startup registers zero resource types, and recovery requires fixing the file AND restarting — hot-load only helps a connector that started successfully. Also documented in README and connector.mdx. - Rebuild the hot-load tests on the real SDK entry point (connectorbuilder.NewConnector) and server RPCs so the SDK dictates the lifecycle order instead of the tests assuming it; the previous helper called Validate before ResourceSyncers — the reverse of production — leaving the actual startup path untested. Add TestHotReload_InvalidFileAtStartupRequiresRestart pinning the startup-failure behavior. Co-Authored-By: Claude Fable 5 --- README.md | 2 +- docs/connector.mdx | 2 +- pkg/connector/connector.go | 98 ++++++++++----- pkg/connector/hot_reload_test.go | 198 ++++++++++++++++++++----------- 4 files changed, 196 insertions(+), 104 deletions(-) diff --git a/README.md b/README.md index a66f8bd7..ed7e5372 100644 --- a/README.md +++ b/README.md @@ -41,7 +41,7 @@ baton-file -i data.xlsx ## Hot-Load (File Change Detection) -When baton-file runs as a long-lived service, edits to the input file are picked up automatically at the start of the next sync cycle — no restart required. This applies to the **data** sections: users, resources, entitlements, grants, and inheritance mappings. **Schema changes** — introducing a brand-new resource type, or changing an existing type's trait — still require a service restart, because the SDK registers resource types and their traits once at process startup. If the edited file fails validation, the sync fails with the validation error and the connector keeps serving the last successfully loaded data. If the file changes while a sync is in flight (for example via health-check revalidation), affected listings restart safely against the new contents instead of resuming into stale page offsets. +When baton-file runs as a long-lived service, edits to the input file are picked up automatically at the start of the next sync cycle — no restart required. This applies to the **data** sections: users, resources, entitlements, grants, and inheritance mappings. **Schema changes** — introducing a brand-new resource type, or changing an existing type's trait — still require a service restart, because the SDK registers resource types and their traits once at process startup. If the edited file fails validation, the sync fails with the validation error and the connector keeps serving the last successfully loaded data. If the file changes while a sync is in flight (for example via health-check revalidation), affected listings restart safely against the new contents instead of resuming into stale page offsets. Hot-load requires a service that started successfully: if the input file is invalid when the service first starts, no resource types get registered and syncs fail until the file is fixed **and the service is restarted**. **IMPORTANT (maintainers and AI agents):** this behavior is a deliberate contract, not an accident of implementation. The SDK calls `ResourceSyncers()` exactly once per process, so `Validate()` — which runs at the start of every sync — is the only per-sync hook and is where the file is re-read and the shared cache is republished (see `cacheHolder` in `pkg/connector/connector.go`). Do not move the file load to construction time, capture cache snapshots in builders, or assume the SDK refreshes data between syncs (it does not — that assumption caused a regression once already). `TestHotReload_DataChangesPickedUpBySync` in `pkg/connector/hot_reload_test.go` enforces this contract; if it fails after your change, the change is wrong, not the test. diff --git a/docs/connector.mdx b/docs/connector.mdx index f0c99d81..622064bb 100644 --- a/docs/connector.mdx +++ b/docs/connector.mdx @@ -34,7 +34,7 @@ Resource types are defined by the input file (via each row's `trait`), so the ex - **Read-only.** The connector cannot create, modify, revoke, or deprovision access in the source system — it only reads what the file describes. - **Full sync only.** Every sync reads the entire file; there is no incremental/targeted sync. - **File-defined data.** C1 only knows what the file contains. Resource IDs must be globally unique across users and resources. -- **File edits apply on the next sync.** When the connector runs as a long-lived service, changes to the file's data — users, resources, entitlements, grants, and inheritance mappings — are picked up automatically at the start of the next sync, with no restart needed. Adding a new resource type, or changing an existing type's trait, does require restarting the connector. If an edited file fails validation, the sync fails with the validation error and the connector keeps serving the last successfully loaded data. +- **File edits apply on the next sync.** When the connector runs as a long-lived service, changes to the file's data — users, resources, entitlements, grants, and inheritance mappings — are picked up automatically at the start of the next sync, with no restart needed. Adding a new resource type, or changing an existing type's trait, does require restarting the connector. If an edited file fails validation, the sync fails with the validation error and the connector keeps serving the last successfully loaded data. If the file is invalid when the connector first starts (rather than edited mid-run), fix the file and restart the connector. ## Gather File connector credentials diff --git a/pkg/connector/connector.go b/pkg/connector/connector.go index 4e9a5587..e7c97bdc 100644 --- a/pkg/connector/connector.go +++ b/pkg/connector/connector.go @@ -8,6 +8,7 @@ import ( "os" "sort" "strings" + "sync" "sync/atomic" "github.com/conductorone/baton-file/pkg/client" @@ -28,6 +29,38 @@ import ( type FileConnector struct { inputFilePath string cache cacheHolder + + // refreshMu serializes the read → build → publish sequence (refresh). + // The holder's atomic pointer keeps readers lock-free, but without this + // lock two concurrent Validate() calls (a health probe racing a sync + // start across two file edits) could publish out of order and leave the + // holder one generation stale until the next Validate(). A + // CompareAndSwap loop cannot fix that: generations are content hashes + // with no ordering, so on CAS failure there is no way to tell which + // build is newer. Serializing the whole sequence makes publish order + // follow file-read order, and also prevents two full cache builds from + // running concurrently (which would triple peak memory). + refreshMu sync.Mutex +} + +// refresh re-reads the input file and publishes a fresh cache when its +// contents changed, returning the cache now being served and whether it was +// republished. On error nothing is stored, so the previously published cache +// keeps serving last-known-good data. +func (fc *FileConnector) refresh(ctx context.Context) (*syncCache, bool, error) { + fc.refreshMu.Lock() + defer fc.refreshMu.Unlock() + + previous := fc.cache.load() + cache, err := loadValidatedCache(ctx, fc.inputFilePath, previous) + if err != nil { + return nil, false, err + } + if cache == previous { + return cache, false, nil + } + fc.cache.store(cache) + return cache, true, nil } // cacheHolder shares the live syncCache between the FileConnector and every @@ -41,11 +74,10 @@ type FileConnector struct { // where the input file is re-read and the fresh syncCache is published here. // Data-section changes to the file MUST be picked up on the next sync without // a restart; only schema changes (a new resource type, or a trait change to -// an existing type) require one. Do NOT -// move the file load back to construction time, capture a *syncCache snapshot -// in a builder, or assume the SDK refreshes anything between syncs (it does -// not — that assumption caused the original hot-load regression in PR #40). -// TestHotReload_DataChangesPickedUpBySync enforces this contract. +// an existing type) require one. Do NOT capture a *syncCache snapshot in a +// builder, and do NOT assume the SDK refreshes anything between syncs (it +// does not — that assumption caused the original hot-load regression in +// PR #40). TestHotReload_DataChangesPickedUpBySync enforces this contract. // // This is deliberate instance state, a documented deviation from the skills' // stateless-connector rule: that rule assumes a remote API as the source of @@ -67,6 +99,13 @@ type FileConnector struct { // sync-lifecycle hook exists. The fingerprint short-circuit in // loadValidatedCache confines this to actual content changes — unchanged // files never swap. +// +// Known limit: hot-load requires a successful start. ResourceSyncers() +// performs the first load at construction; if the file is invalid at that +// moment, zero resource types are registered for the process lifetime and +// every sync fails until the file is fixed AND the process restarts (see +// ResourceSyncers). TestHotReload_InvalidFileAtStartupRequiresRestart pins +// this behavior. type cacheHolder struct { p atomic.Pointer[syncCache] } @@ -172,16 +211,12 @@ func (fc *FileConnector) Validate(ctx context.Context) (annotations.Annotations, // Re-read, validate, and publish the file's current contents. Validate() // runs at the start of every sync, so this is the hot-load refresh point - // (see cacheHolder). On error we return without storing, so the - // previously published cache keeps serving last-known-good data. - previous := fc.cache.load() - cache, err := loadValidatedCache(ctx, fc.inputFilePath, previous) + // (see cacheHolder). + cache, changed, err := fc.refresh(ctx) if err != nil { return nil, err } - if cache != previous { - fc.cache.store(cache) - + if changed { // Debug, not Info: this fires on every sync AND every health-check // probe, so Info would flood default logs on probed deployments. ctxzap.Extract(ctx).Debug("baton-file: refreshed data from input file", @@ -252,27 +287,28 @@ func loadValidatedCache(ctx context.Context, inputFilePath string, current *sync func (fc *FileConnector) ResourceSyncers(ctx context.Context) []connectorbuilder.ResourceSyncerV2 { l := ctxzap.Extract(ctx) - // IMPORTANT: the SDK calls this exactly ONCE per process, at connector - // construction, to register resource types — never again. Only the schema - // (the set of syncers) is fixed here; the data each builder serves flows - // through the shared fc.cache holder so Validate() can refresh it every - // sync. See cacheHolder for the hot-load contract. - cache := fc.cache.load() - if cache == nil { - // Validate() normally runs first and publishes the cache; this - // fallback covers entry points that skip Validate(). This method's - // signature cannot return an error, so log-and-return-nil is - // intentional — the connector stays alive for the next sync cycle. + // IMPORTANT: the SDK calls this exactly ONCE per process, inside + // connectorbuilder.NewConnector at construction (vendor/.../ + // connectorbuilder/connectorbuilder.go:216), BEFORE any Validate() — + // this is the FIRST load of the file, not a fallback. The set of + // resource types registered here is fixed for the process lifetime; + // Validate() then refreshes the data behind fc.cache on every sync. See + // cacheHolder for the hot-load contract. + cache, _, err := fc.refresh(ctx) + if err != nil { + // The signature cannot return an error, so log-and-return-nil is + // all that is possible — and the consequence is severe: with zero + // syncers registered, every sync fails with FailedPrecondition + // ("no resource builders found", vendor/.../connectorbuilder/ + // resource_syncer.go:102) even after the operator fixes the file. + // Recovering from a bad file AT STARTUP requires a process restart; + // hot-load only helps a connector that started with a valid file. // Warn, not Error: Validate() returns the real error to the SDK on // every sync, so this log is a secondary signal, and house rules - // classify input/config failures as Warn. - var err error - cache, err = loadValidatedCache(ctx, fc.inputFilePath, nil) - if err != nil { - l.Warn("baton-file: failed to load input file", zap.Error(err)) - return nil - } - fc.cache.store(cache) + // classify input failures as Warn. + l.Warn("baton-file: failed to load input file at startup; syncs will fail until the file is fixed and the service is restarted", + zap.Error(err)) + return nil } var syncers []connectorbuilder.ResourceSyncerV2 diff --git a/pkg/connector/hot_reload_test.go b/pkg/connector/hot_reload_test.go index ad733fdc..2eb00ab9 100644 --- a/pkg/connector/hot_reload_test.go +++ b/pkg/connector/hot_reload_test.go @@ -7,23 +7,26 @@ import ( "testing" v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" - "github.com/conductorone/baton-sdk/pkg/pagination" - rs "github.com/conductorone/baton-sdk/pkg/types/resource" + "github.com/conductorone/baton-sdk/pkg/connectorbuilder" + "github.com/conductorone/baton-sdk/pkg/types" "github.com/stretchr/testify/require" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) // Hot-load contract tests (see cacheHolder in connector.go). // -// The SDK constructs the connector and calls ResourceSyncers() exactly once -// per process, then re-runs Validate() at the start of every sync. In a -// long-running service, data-section changes to the input file MUST be -// served on the next sync without a restart. These tests mirror that -// lifecycle: one ResourceSyncers() call, then file edits followed by -// Validate(), asserting through the ORIGINAL builders. +// Construction goes through the real SDK entry point, +// connectorbuilder.NewConnector, so the SDK itself dictates the lifecycle: +// ResourceSyncers() runs ONCE during construction (the first file load), +// and Validate() runs afterwards, once per sync. The call order is not +// hand-assembled in these tests precisely so that a wrong assumption about +// it cannot produce passing tests. Assertions go through the same server +// RPCs the SDK syncer uses (ListResources, ListGrants, Validate). // // Unlike the other connector tests, fixtures here are real files written to -// disk rather than in-memory LoadedData: hot-load IS the file → Validate → -// cache path, so bypassing the loaders would test nothing. +// disk rather than in-memory LoadedData: hot-load IS the +// file → refresh → cache path, so bypassing the loaders would test nothing. const hotReloadCSVBase = `record_type,id,display_name,email,status,type,profile.department,profile.title,resource_type,trait,description,resource_id,entitlement_slug,principal_id user,alex.taylor,Alex Taylor,alex.taylor@example.com,enabled,human,Engineering,Software Engineer,,,,,, @@ -45,92 +48,127 @@ const hotReloadCSVInvalid = hotReloadCSVBase + `user,alex.taylor,Alex Duplicate,alex.duplicate@example.com,enabled,human,,,,,,,, ` -// startConnector writes the CSV, then walks the once-per-process SDK startup -// sequence (Validate, then ResourceSyncers) and returns the connector plus -// the long-lived builders keyed by resource type id. -func startConnector(t *testing.T, path, csv string) (*FileConnector, map[string]*resourceBuilder) { +// startServer writes the CSV and constructs the connector through the real +// SDK entry point, mirroring a service-mode process start. +func startServer(t *testing.T, path, csv string) (types.ConnectorServer, *FileConnector) { t.Helper() - ctx := context.Background() require.NoError(t, os.WriteFile(path, []byte(csv), 0o600)) - fc := &FileConnector{inputFilePath: path} - _, err := fc.Validate(ctx) + server, err := connectorbuilder.NewConnector(context.Background(), fc) require.NoError(t, err) + return server, fc +} + +func serverValidate(t *testing.T, server types.ConnectorServer) error { + t.Helper() + _, err := server.Validate(context.Background(), v2.ConnectorServiceValidateRequest_builder{}.Build()) + return err +} + +// listAll pages through ListResources for one resource type, with pageSize +// controlling how many resources each RPC returns (0 = server default). +func listAll(t *testing.T, server types.ConnectorServer, typeID string, pageSize int) []*v2.Resource { + t.Helper() + ctx := context.Background() + var out []*v2.Resource + token := "" + for { + resp, err := server.ListResources(ctx, v2.ResourcesServiceListResourcesRequest_builder{ + ResourceTypeId: typeID, + PageToken: token, + PageSize: uint32(pageSize), //nolint:gosec // test-controlled small values + }.Build()) + require.NoError(t, err) + out = append(out, resp.GetList()...) + if token = resp.GetNextPageToken(); token == "" { + return out + } + } +} - builders := make(map[string]*resourceBuilder) - for _, s := range fc.ResourceSyncers(ctx) { - b, ok := s.(*resourceBuilder) - require.True(t, ok) - builders[b.ResourceType(ctx).GetId()] = b +func listGrants(t *testing.T, server types.ConnectorServer, res *v2.Resource) []*v2.Grant { + t.Helper() + ctx := context.Background() + var out []*v2.Grant + token := "" + for { + resp, err := server.ListGrants(ctx, v2.GrantsServiceListGrantsRequest_builder{ + Resource: res, + PageToken: token, + }.Build()) + require.NoError(t, err) + out = append(out, resp.GetList()...) + if token = resp.GetNextPageToken(); token == "" { + return out + } } - require.Contains(t, builders, "user") - require.Contains(t, builders, "team") - return fc, builders } -func findResource(t *testing.T, b *resourceBuilder, id string) *v2.Resource { +func findResource(t *testing.T, server types.ConnectorServer, typeID, id string) *v2.Resource { t.Helper() - for _, r := range allResources(t, b, nil) { + for _, r := range listAll(t, server, typeID, 0) { if r.GetId().GetResource() == id { return r } } - t.Fatalf("resource %q not found", id) + t.Fatalf("resource %s/%s not found", typeID, id) return nil } func TestHotReload_DataChangesPickedUpBySync(t *testing.T) { - ctx := context.Background() path := filepath.Join(t.TempDir(), "input.csv") - fc, builders := startConnector(t, path, hotReloadCSVBase) + server, _ := startServer(t, path, hotReloadCSVBase) - require.Len(t, allResources(t, builders["user"], nil), 2) - engineering := findResource(t, builders["team"], "engineering") - require.Len(t, allGrants(t, builders["team"], engineering), 1) + require.Len(t, listAll(t, server, "user", 0), 2) + engineering := findResource(t, server, "team", "engineering") + require.Len(t, listGrants(t, server, engineering), 1) // The customer edits the file while the service keeps running. require.NoError(t, os.WriteFile(path, []byte(hotReloadCSVUpdated), 0o600)) - // Validate() is the only per-sync hook the SDK gives this connector, so - // it alone must refresh the data the original builders serve. - _, err := fc.Validate(ctx) - require.NoError(t, err) + // Validate() is the per-sync hook the SDK runs at the start of every + // sync; it alone must refresh the data the registered syncers serve. + require.NoError(t, serverValidate(t, server)) - require.Len(t, allResources(t, builders["user"], nil), 3) - require.Len(t, allGrants(t, builders["team"], engineering), 2) + require.Len(t, listAll(t, server, "user", 0), 3) + require.Len(t, listGrants(t, server, engineering), 2) } func TestHotReload_SwapMidListingRestartsPagination(t *testing.T) { ctx := context.Background() path := filepath.Join(t.TempDir(), "input.csv") - fc, builders := startConnector(t, path, hotReloadCSVBase) + server, _ := startServer(t, path, hotReloadCSVBase) // Fetch the first page (1 of 2 users) so a token is in flight. - firstPage, results, err := builders["user"].List(ctx, nil, - rs.SyncOpAttrs{PageToken: pagination.Token{Size: 1}}) + first, err := server.ListResources(ctx, v2.ResourcesServiceListResourcesRequest_builder{ + ResourceTypeId: "user", + PageSize: 1, + }.Build()) require.NoError(t, err) - require.Len(t, firstPage, 1) - require.NotEmpty(t, results.NextPageToken) + require.Len(t, first.GetList(), 1) + require.NotEmpty(t, first.GetNextPageToken()) - // The file changes and a Validate (sync start or health probe) swaps the - // cache while the listing is in flight. + // The file changes and a Validate (sync start or health probe) swaps + // the cache while the listing is in flight. require.NoError(t, os.WriteFile(path, []byte(hotReloadCSVUpdated), 0o600)) - _, err = fc.Validate(ctx) - require.NoError(t, err) + require.NoError(t, serverValidate(t, server)) - // Resuming with the stale token must restart the listing against the new - // generation — every current user is returned exactly once, rather than - // offsets silently skipping or duplicating rows across generations. + // Resuming with the stale token must restart the listing against the + // new generation — every current user is returned exactly once, rather + // than offsets silently skipping or duplicating rows across generations. seen := map[string]int{} - token := results.NextPageToken + token := first.GetNextPageToken() for { - page, res, err := builders["user"].List(ctx, nil, - rs.SyncOpAttrs{PageToken: pagination.Token{Token: token, Size: 1}}) + resp, err := server.ListResources(ctx, v2.ResourcesServiceListResourcesRequest_builder{ + ResourceTypeId: "user", + PageToken: token, + PageSize: 1, + }.Build()) require.NoError(t, err) - for _, r := range page { + for _, r := range resp.GetList() { seen[r.GetId().GetResource()]++ } - if token = res.NextPageToken; token == "" { + if token = resp.GetNextPageToken(); token == "" { break } } @@ -138,39 +176,57 @@ func TestHotReload_SwapMidListingRestartsPagination(t *testing.T) { } func TestHotReload_UnchangedFileSkipsRebuildAndSwap(t *testing.T) { - ctx := context.Background() path := filepath.Join(t.TempDir(), "input.csv") - fc, _ := startConnector(t, path, hotReloadCSVBase) + server, fc := startServer(t, path, hotReloadCSVBase) // A Validate against unchanged content (every health probe, and every - // sync where nobody edited the file) must return the already-published - // cache: no rebuild, and critically no pointer swap under an in-flight - // sync. Pointer identity is the contract. + // sync where nobody edited the file) must keep serving the cache the + // construction-time load published: no rebuild, and critically no + // pointer swap under an in-flight sync. Pointer identity is the + // contract. before := fc.cache.load() - _, err := fc.Validate(ctx) - require.NoError(t, err) + require.NotNil(t, before) + require.NoError(t, serverValidate(t, server)) require.Same(t, before, fc.cache.load()) // A real change still swaps. require.NoError(t, os.WriteFile(path, []byte(hotReloadCSVUpdated), 0o600)) - _, err = fc.Validate(ctx) - require.NoError(t, err) + require.NoError(t, serverValidate(t, server)) require.NotSame(t, before, fc.cache.load()) } func TestHotReload_InvalidFileKeepsLastGoodCache(t *testing.T) { - ctx := context.Background() path := filepath.Join(t.TempDir(), "input.csv") - fc, builders := startConnector(t, path, hotReloadCSVBase) + server, _ := startServer(t, path, hotReloadCSVBase) - require.Len(t, allResources(t, builders["user"], nil), 2) + require.Len(t, listAll(t, server, "user", 0), 2) // The file is replaced with one that fails validation: the sync must // fail loudly while the last-known-good data keeps being served. require.NoError(t, os.WriteFile(path, []byte(hotReloadCSVInvalid), 0o600)) + require.Error(t, serverValidate(t, server)) + + require.Len(t, listAll(t, server, "user", 0), 2) +} + +func TestHotReload_InvalidFileAtStartupRequiresRestart(t *testing.T) { + // Pins the known limit documented on cacheHolder: hot-load cannot + // recover a process that STARTED with an invalid file. ResourceSyncers() + // runs once at construction; on load failure it registers zero types, + // and that registration never happens again. + ctx := context.Background() + path := filepath.Join(t.TempDir(), "input.csv") + server, _ := startServer(t, path, hotReloadCSVInvalid) + + _, err := server.ListResourceTypes(ctx, v2.ResourceTypesServiceListResourceTypesRequest_builder{}.Build()) + require.Equal(t, codes.FailedPrecondition, status.Code(err)) - _, err := fc.Validate(ctx) - require.Error(t, err) + // The operator fixes the file. Validate now succeeds (and publishes a + // good cache), but the type registration is gone for the process + // lifetime — syncs still fail. Only a restart recovers. + require.NoError(t, os.WriteFile(path, []byte(hotReloadCSVBase), 0o600)) + require.NoError(t, serverValidate(t, server)) - require.Len(t, allResources(t, builders["user"], nil), 2) + _, err = server.ListResourceTypes(ctx, v2.ResourceTypesServiceListResourceTypesRequest_builder{}.Build()) + require.Equal(t, codes.FailedPrecondition, status.Code(err)) } From 24e37f776e2c5a6350a2370051c788ba2ab5709d Mon Sep 17 00:00:00 2001 From: Matthew Scobell Date: Thu, 13 Aug 2026 21:31:55 -0400 Subject: [PATCH 06/11] docs: align README hot-load contract with the corrected lifecycle The in-code contract was corrected to reflect that ResourceSyncers() performs the first file load at construction; the README paragraph mirroring it still said "do not move the file load to construction time", contradicting the code it points to. Reword to the actual invariant: the construction-time load is the first load, never the only one. Co-Authored-By: Claude Fable 5 --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index ed7e5372..6eb53a9f 100644 --- a/README.md +++ b/README.md @@ -43,7 +43,7 @@ baton-file -i data.xlsx When baton-file runs as a long-lived service, edits to the input file are picked up automatically at the start of the next sync cycle — no restart required. This applies to the **data** sections: users, resources, entitlements, grants, and inheritance mappings. **Schema changes** — introducing a brand-new resource type, or changing an existing type's trait — still require a service restart, because the SDK registers resource types and their traits once at process startup. If the edited file fails validation, the sync fails with the validation error and the connector keeps serving the last successfully loaded data. If the file changes while a sync is in flight (for example via health-check revalidation), affected listings restart safely against the new contents instead of resuming into stale page offsets. Hot-load requires a service that started successfully: if the input file is invalid when the service first starts, no resource types get registered and syncs fail until the file is fixed **and the service is restarted**. -**IMPORTANT (maintainers and AI agents):** this behavior is a deliberate contract, not an accident of implementation. The SDK calls `ResourceSyncers()` exactly once per process, so `Validate()` — which runs at the start of every sync — is the only per-sync hook and is where the file is re-read and the shared cache is republished (see `cacheHolder` in `pkg/connector/connector.go`). Do not move the file load to construction time, capture cache snapshots in builders, or assume the SDK refreshes data between syncs (it does not — that assumption caused a regression once already). `TestHotReload_DataChangesPickedUpBySync` in `pkg/connector/hot_reload_test.go` enforces this contract; if it fails after your change, the change is wrong, not the test. +**IMPORTANT (maintainers and AI agents):** this behavior is a deliberate contract, not an accident of implementation. The SDK calls `ResourceSyncers()` exactly once per process, so `Validate()` — which runs at the start of every sync — is the only per-sync hook and is where the file is re-read and the shared cache is republished (see `cacheHolder` in `pkg/connector/connector.go`). The construction-time load in `ResourceSyncers()` is the *first* load, never the only one: do not capture cache snapshots in builders, and do not assume the SDK refreshes data between syncs (it does not — that assumption caused a regression once already). `TestHotReload_DataChangesPickedUpBySync` in `pkg/connector/hot_reload_test.go` enforces this contract; if it fails after your change, the change is wrong, not the test. ## Templates From 7ec33272758b698a29bcdf41c6af328c871926ce Mon Sep 17 00:00:00 2001 From: Matthew Scobell Date: Thu, 13 Aug 2026 23:15:53 -0400 Subject: [PATCH 07/11] fix: treat a valid data-less file as a legitimate empty source of truth MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A file with schema but no data rows previously registered zero resource types, so every sync failed with FailedPrecondition for the process lifetime — the same dead-end as an invalid file. Per product intent an empty file is a valid state: ResourceSyncers now registers the standard trait types (the same set StaticCapabilitiesConnector declares as the connector's capabilities), so syncs succeed, emit nothing, and data rows added later hot-load without a restart. Rows using custom type IDs remain a schema change requiring restart; Validate() now warns when a refresh publishes types that were not registered at startup — previously that drift was completely silent. The invalid-file-at-startup trap is unchanged and still pinned by its test; new tests pin empty-start sync + hot-load and the custom-type restart contract. README and connector.mdx updated to match. Co-Authored-By: Claude Fable 5 --- README.md | 2 +- docs/connector.mdx | 2 +- pkg/connector/connector.go | 86 ++++++++++++++++++++++++++------ pkg/connector/hot_reload_test.go | 59 ++++++++++++++++++++++ 4 files changed, 132 insertions(+), 17 deletions(-) diff --git a/README.md b/README.md index 6eb53a9f..80df5189 100644 --- a/README.md +++ b/README.md @@ -41,7 +41,7 @@ baton-file -i data.xlsx ## Hot-Load (File Change Detection) -When baton-file runs as a long-lived service, edits to the input file are picked up automatically at the start of the next sync cycle — no restart required. This applies to the **data** sections: users, resources, entitlements, grants, and inheritance mappings. **Schema changes** — introducing a brand-new resource type, or changing an existing type's trait — still require a service restart, because the SDK registers resource types and their traits once at process startup. If the edited file fails validation, the sync fails with the validation error and the connector keeps serving the last successfully loaded data. If the file changes while a sync is in flight (for example via health-check revalidation), affected listings restart safely against the new contents instead of resuming into stale page offsets. Hot-load requires a service that started successfully: if the input file is invalid when the service first starts, no resource types get registered and syncs fail until the file is fixed **and the service is restarted**. +When baton-file runs as a long-lived service, edits to the input file are picked up automatically at the start of the next sync cycle — no restart required. This applies to the **data** sections: users, resources, entitlements, grants, and inheritance mappings. **Schema changes** — introducing a brand-new resource type, or changing an existing type's trait — still require a service restart, because the SDK registers resource types and their traits once at process startup. If the edited file fails validation, the sync fails with the validation error and the connector keeps serving the last successfully loaded data. If the file changes while a sync is in flight (for example via health-check revalidation), affected listings restart safely against the new contents instead of resuming into stale page offsets. A file that is valid but has no data rows is a legitimate empty state: the connector registers its standard resource types (user, group, role, app, secret), syncs empty, and hot-loads data rows as they are added — though rows using custom resource type IDs still require a restart, like any schema change. Hot-load does require a service that started successfully: if the input file is **invalid** when the service first starts, no resource types get registered and syncs fail until the file is fixed **and the service is restarted**. **IMPORTANT (maintainers and AI agents):** this behavior is a deliberate contract, not an accident of implementation. The SDK calls `ResourceSyncers()` exactly once per process, so `Validate()` — which runs at the start of every sync — is the only per-sync hook and is where the file is re-read and the shared cache is republished (see `cacheHolder` in `pkg/connector/connector.go`). The construction-time load in `ResourceSyncers()` is the *first* load, never the only one: do not capture cache snapshots in builders, and do not assume the SDK refreshes data between syncs (it does not — that assumption caused a regression once already). `TestHotReload_DataChangesPickedUpBySync` in `pkg/connector/hot_reload_test.go` enforces this contract; if it fails after your change, the change is wrong, not the test. diff --git a/docs/connector.mdx b/docs/connector.mdx index 622064bb..a1ecf3b3 100644 --- a/docs/connector.mdx +++ b/docs/connector.mdx @@ -34,7 +34,7 @@ Resource types are defined by the input file (via each row's `trait`), so the ex - **Read-only.** The connector cannot create, modify, revoke, or deprovision access in the source system — it only reads what the file describes. - **Full sync only.** Every sync reads the entire file; there is no incremental/targeted sync. - **File-defined data.** C1 only knows what the file contains. Resource IDs must be globally unique across users and resources. -- **File edits apply on the next sync.** When the connector runs as a long-lived service, changes to the file's data — users, resources, entitlements, grants, and inheritance mappings — are picked up automatically at the start of the next sync, with no restart needed. Adding a new resource type, or changing an existing type's trait, does require restarting the connector. If an edited file fails validation, the sync fails with the validation error and the connector keeps serving the last successfully loaded data. If the file is invalid when the connector first starts (rather than edited mid-run), fix the file and restart the connector. +- **File edits apply on the next sync.** When the connector runs as a long-lived service, changes to the file's data — users, resources, entitlements, grants, and inheritance mappings — are picked up automatically at the start of the next sync, with no restart needed. Adding a new resource type, or changing an existing type's trait, does require restarting the connector. If an edited file fails validation, the sync fails with the validation error and the connector keeps serving the last successfully loaded data. A file with no data rows is valid: the connector syncs empty and picks up rows as they are added. If the file is invalid when the connector first starts (rather than edited mid-run), fix the file and restart the connector. ## Gather File connector credentials diff --git a/pkg/connector/connector.go b/pkg/connector/connector.go index e7c97bdc..069c59a5 100644 --- a/pkg/connector/connector.go +++ b/pkg/connector/connector.go @@ -41,6 +41,14 @@ type FileConnector struct { // follow file-read order, and also prevents two full cache builds from // running concurrently (which would triple peak memory). refreshMu sync.Mutex + + // registeredTypes records the resource type IDs registered by the + // construction-time ResourceSyncers() call. Written once during + // construction (before the gRPC server starts serving) and read-only + // afterwards, so it needs no lock. Validate() uses it to warn when a + // file edit introduces types that have no registered syncer and + // therefore will not sync until restart. + registeredTypes map[string]struct{} } // refresh re-reads the input file and publishes a fresh cache when its @@ -101,11 +109,14 @@ func (fc *FileConnector) refresh(ctx context.Context) (*syncCache, bool, error) // files never swap. // // Known limit: hot-load requires a successful start. ResourceSyncers() -// performs the first load at construction; if the file is invalid at that +// performs the first load at construction; if the file is INVALID at that // moment, zero resource types are registered for the process lifetime and // every sync fails until the file is fixed AND the process restarts (see // ResourceSyncers). TestHotReload_InvalidFileAtStartupRequiresRestart pins -// this behavior. +// this behavior. A file that is valid but has no data rows is NOT this trap: +// it is a legitimate empty source of truth — ResourceSyncers registers the +// standard trait types so syncs succeed (empty) and later data rows +// hot-load; TestHotReload_EmptyFileAtStartupSyncsAndHotLoads pins that. type cacheHolder struct { p atomic.Pointer[syncCache] } @@ -222,6 +233,26 @@ func (fc *FileConnector) Validate(ctx context.Context) (annotations.Annotations, ctxzap.Extract(ctx).Debug("baton-file: refreshed data from input file", zap.Int("resource_types", len(cache.resourceTypes)), zap.Int("resources", len(cache.resources))) + + // Schema drift is otherwise silent: rows whose resource type was not + // registered at startup simply never sync. Warn so the operator + // knows a restart is needed for them. registeredTypes is nil only + // when ResourceSyncers() has not run (unit tests calling Validate + // directly); skip the check there. + if fc.registeredTypes != nil { + var missing []string + for typeID := range cache.resourceTypes { + if _, ok := fc.registeredTypes[typeID]; !ok { + missing = append(missing, typeID) + } + } + if len(missing) > 0 { + sort.Strings(missing) + ctxzap.Extract(ctx).Warn( + "baton-file: file contains resource types not registered at startup; restart the service to sync them", + zap.Strings("resource_types", missing)) + } + } } return nil, nil @@ -312,19 +343,44 @@ func (fc *FileConnector) ResourceSyncers(ctx context.Context) []connectorbuilder } var syncers []connectorbuilder.ResourceSyncerV2 - for _, rt := range cache.resourceTypes { - // resourceType is INTENTIONALLY frozen at registration time and never - // refreshed from later cache generations. Hot-load covers the file's - // data sections only; schema — the set of resource types and their - // traits — is fixed for the process lifetime because the SDK registers - // syncers by type exactly once. Editing an existing type's trait in - // the file therefore requires a restart, same as adding a new type; - // resolving rt from the live cache here would not change that, since - // the SDK-side registration would still hold the startup-time type. - syncers = append(syncers, &resourceBuilder{cache: &fc.cache, resourceType: rt}) - } - - l.Info("baton-file: created resource syncers", zap.Int("count", len(syncers))) + if len(cache.resourceTypes) == 0 { + // A valid file with no data rows is a legitimate state — an empty + // source of truth — not an error. Register the standard trait types + // (the same set StaticCapabilitiesConnector declares as this + // connector's capabilities) so syncs succeed, emit nothing, and + // data rows added to the file later hot-load without a restart. + // Rows using CUSTOM type IDs still need a restart, like any schema + // change; Validate() warns when it sees them. + for name := range TraitMap { + syncers = append(syncers, &resourceBuilder{ + cache: &fc.cache, + resourceType: buildDynamicResourceType(name, name), + }) + } + l.Info("baton-file: input file has no data rows; registered standard resource types", + zap.Int("count", len(syncers))) + } else { + for _, rt := range cache.resourceTypes { + // resourceType is INTENTIONALLY frozen at registration time and + // never refreshed from later cache generations. Hot-load covers + // the file's data sections only; schema — the set of resource + // types and their traits — is fixed for the process lifetime + // because the SDK registers syncers by type exactly once. + // Editing an existing type's trait in the file therefore + // requires a restart, same as adding a new type; resolving rt + // from the live cache here would not change that, since the + // SDK-side registration would still hold the startup-time type. + syncers = append(syncers, &resourceBuilder{cache: &fc.cache, resourceType: rt}) + } + l.Info("baton-file: created resource syncers", zap.Int("count", len(syncers))) + } + + fc.registeredTypes = make(map[string]struct{}, len(syncers)) + for _, s := range syncers { + if rb, ok := s.(*resourceBuilder); ok { + fc.registeredTypes[rb.resourceType.GetId()] = struct{}{} + } + } return syncers } diff --git a/pkg/connector/hot_reload_test.go b/pkg/connector/hot_reload_test.go index 2eb00ab9..21b054d7 100644 --- a/pkg/connector/hot_reload_test.go +++ b/pkg/connector/hot_reload_test.go @@ -48,6 +48,21 @@ const hotReloadCSVInvalid = hotReloadCSVBase + `user,alex.taylor,Alex Duplicate,alex.duplicate@example.com,enabled,human,,,,,,,, ` +// hotReloadCSVHeaderOnly is a valid file with schema but zero data rows — a +// legitimate empty source of truth, not an error. +const hotReloadCSVHeaderOnly = `record_type,id,display_name,email,status,type,profile.department,profile.title,resource_type,trait,description,resource_id,entitlement_slug,principal_id +` + +// hotReloadCSVStandardTypes uses only type IDs within the standard trait set +// (user, group) that an empty-start connector registers. +const hotReloadCSVStandardTypes = hotReloadCSVHeaderOnly + + `user,alex.taylor,Alex Taylor,alex.taylor@example.com,enabled,human,Engineering,Software Engineer,,,,,, +user,sam.johnson,Sam Johnson,sam.johnson@example.com,enabled,human,Marketing,Marketing Manager,,,,,, +resource,platform,Platform,,,,,,group,group,Platform group,,, +entitlement,,Member,,,,,,,,,platform,member, +direct_user_grant,,,,,,,,,,,platform,member,alex.taylor +` + // startServer writes the CSV and constructs the connector through the real // SDK entry point, mirroring a service-mode process start. func startServer(t *testing.T, path, csv string) (types.ConnectorServer, *FileConnector) { @@ -209,6 +224,50 @@ func TestHotReload_InvalidFileKeepsLastGoodCache(t *testing.T) { require.Len(t, listAll(t, server, "user", 0), 2) } +func TestHotReload_EmptyFileAtStartupSyncsAndHotLoads(t *testing.T) { + // A valid file with no data rows must work: the connector registers the + // standard trait types (matching its declared capabilities), syncs + // empty, and hot-loads data rows added later. + ctx := context.Background() + path := filepath.Join(t.TempDir(), "input.csv") + server, _ := startServer(t, path, hotReloadCSVHeaderOnly) + + rts, err := server.ListResourceTypes(ctx, v2.ResourceTypesServiceListResourceTypesRequest_builder{}.Build()) + require.NoError(t, err) + require.Len(t, rts.GetList(), len(TraitMap)) + require.Empty(t, listAll(t, server, "user", 0)) + require.Empty(t, listAll(t, server, "group", 0)) + + // Data rows using standard type IDs appear on the next sync. + require.NoError(t, os.WriteFile(path, []byte(hotReloadCSVStandardTypes), 0o600)) + require.NoError(t, serverValidate(t, server)) + + require.Len(t, listAll(t, server, "user", 0), 2) + platform := findResource(t, server, "group", "platform") + require.Len(t, listGrants(t, server, platform), 1) +} + +func TestHotReload_CustomTypeAfterEmptyStartRequiresRestart(t *testing.T) { + // Custom type IDs are schema: an empty-start connector registered only + // the standard trait types, so rows with a custom type (here "team") + // do not sync until restart — the documented schema-change contract. + // Validate() warns about them; this test pins the RPC-visible behavior. + ctx := context.Background() + path := filepath.Join(t.TempDir(), "input.csv") + server, _ := startServer(t, path, hotReloadCSVHeaderOnly) + + require.NoError(t, os.WriteFile(path, []byte(hotReloadCSVBase), 0o600)) + require.NoError(t, serverValidate(t, server)) + + // Users hot-load (standard type); the "team"-typed resource has no + // registered syncer, so listing it fails with NotFound until restart. + require.Len(t, listAll(t, server, "user", 0), 2) + _, err := server.ListResources(ctx, v2.ResourcesServiceListResourcesRequest_builder{ + ResourceTypeId: "team", + }.Build()) + require.Equal(t, codes.NotFound, status.Code(err)) +} + func TestHotReload_InvalidFileAtStartupRequiresRestart(t *testing.T) { // Pins the known limit documented on cacheHolder: hot-load cannot // recover a process that STARTED with an invalid file. ResourceSyncers() From 4497ca17061b91900eb7054426fa6be18bf738a0 Mon Sep 17 00:00:00 2001 From: Matthew Scobell Date: Fri, 14 Aug 2026 01:36:31 -0400 Subject: [PATCH 08/11] fix: level-triggered drift warning; observable generation restarts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - The schema-drift check now runs on EVERY Validate(), not just the one where the file's content hash changed: the condition persists until restart, so an edge-triggered warning meant one missable log line for a permanent problem. Logarithmic sampling (occurrences 1, 2, 4, 8, ... with total_occurrences) keeps probed deployments from flooding; the counter resets when drift resolves so a new episode warns immediately. TestHotReload_SchemaDriftCheckIsLevelTriggered pins the level-triggered contract. - paginate() now logs a Warn when a generation mismatch (or a legacy pre-upgrade token) actually restarts a listing, making mid-sync cache swaps and the documented cross-phase consistency window observable in logs instead of inferable-only. The probe/sync-start distinction the alternative guard would need does not exist connector-side (same RPC, no sync-lifecycle hook) — recorded on the cacheHolder contract. Co-Authored-By: Claude Fable 5 --- pkg/connector/connector.go | 53 +++++++++++++++++++++----------- pkg/connector/hot_reload_test.go | 27 ++++++++++++++++ pkg/connector/resources.go | 26 +++++++++++----- pkg/connector/resources_test.go | 16 +++++----- 4 files changed, 88 insertions(+), 34 deletions(-) diff --git a/pkg/connector/connector.go b/pkg/connector/connector.go index 069c59a5..e06a36a1 100644 --- a/pkg/connector/connector.go +++ b/pkg/connector/connector.go @@ -49,6 +49,14 @@ type FileConnector struct { // file edit introduces types that have no registered syncer and // therefore will not sync until restart. registeredTypes map[string]struct{} + + // driftOccurrences counts consecutive Validate() calls that observed + // unregistered resource types, driving logarithmic log sampling: the + // drift condition persists until restart, so it must warn repeatedly + // (level-triggered, not just on the sync where it appeared), but + // Validate also runs on every health probe, so warning every time + // would flood logs. Reset to zero when the drift resolves. + driftOccurrences atomic.Uint64 } // refresh re-reads the input file and publishes a fresh cache when its @@ -106,7 +114,11 @@ func (fc *FileConnector) refresh(ctx context.Context) (*syncCache, bool, error) // probe Validate and a sync-start Validate arrive as the same RPC, and no // sync-lifecycle hook exists. The fingerprint short-circuit in // loadValidatedCache confines this to actual content changes — unchanged -// files never swap. +// files never swap. Because the window cannot be closed connector-side, it +// is made observable instead: paginate() logs a Warn every time a +// generation mismatch actually restarts a listing, so routine mid-sync +// swaps (e.g. an externally rewritten file under frequent probes) show up +// in logs rather than having to be inferred. // // Known limit: hot-load requires a successful start. ResourceSyncers() // performs the first load at construction; if the file is INVALID at that @@ -233,26 +245,31 @@ func (fc *FileConnector) Validate(ctx context.Context) (annotations.Annotations, ctxzap.Extract(ctx).Debug("baton-file: refreshed data from input file", zap.Int("resource_types", len(cache.resourceTypes)), zap.Int("resources", len(cache.resources))) + } - // Schema drift is otherwise silent: rows whose resource type was not - // registered at startup simply never sync. Warn so the operator - // knows a restart is needed for them. registeredTypes is nil only - // when ResourceSyncers() has not run (unit tests calling Validate - // directly); skip the check there. - if fc.registeredTypes != nil { - var missing []string - for typeID := range cache.resourceTypes { - if _, ok := fc.registeredTypes[typeID]; !ok { - missing = append(missing, typeID) - } - } - if len(missing) > 0 { - sort.Strings(missing) - ctxzap.Extract(ctx).Warn( - "baton-file: file contains resource types not registered at startup; restart the service to sync them", - zap.Strings("resource_types", missing)) + // Schema drift is otherwise silent: rows whose resource type was not + // registered at startup simply never sync. The condition persists until + // restart, so this check is level-triggered — it runs on EVERY + // Validate(), not just the one where the file changed — with logarithmic + // sampling (occurrences 1, 2, 4, 8, ...) so probed deployments are not + // flooded. registeredTypes is nil only when ResourceSyncers() has not + // run (unit tests calling Validate directly); skip the check there. + if fc.registeredTypes != nil { + var missing []string + for typeID := range cache.resourceTypes { + if _, ok := fc.registeredTypes[typeID]; !ok { + missing = append(missing, typeID) } } + if len(missing) == 0 { + fc.driftOccurrences.Store(0) + } else if n := fc.driftOccurrences.Add(1); n&(n-1) == 0 { + sort.Strings(missing) + ctxzap.Extract(ctx).Warn( + "baton-file: file contains resource types not registered at startup; restart the service to sync them", + zap.Strings("resource_types", missing), + zap.Uint64("total_occurrences", n)) + } } return nil, nil diff --git a/pkg/connector/hot_reload_test.go b/pkg/connector/hot_reload_test.go index 21b054d7..38282eed 100644 --- a/pkg/connector/hot_reload_test.go +++ b/pkg/connector/hot_reload_test.go @@ -268,6 +268,33 @@ func TestHotReload_CustomTypeAfterEmptyStartRequiresRestart(t *testing.T) { require.Equal(t, codes.NotFound, status.Code(err)) } +func TestHotReload_SchemaDriftCheckIsLevelTriggered(t *testing.T) { + // Schema drift (file types with no registered syncer) persists until + // restart, so its detection must run on EVERY Validate — including ones + // where the file content is unchanged — not just the sync where the + // drift first appeared. The occurrence counter drives logarithmic log + // sampling (warns at 1, 2, 4, 8, ...) and is the observable contract. + path := filepath.Join(t.TempDir(), "input.csv") + server, fc := startServer(t, path, hotReloadCSVHeaderOnly) + + // Introduce a custom "team" type that the empty start did not register. + require.NoError(t, os.WriteFile(path, []byte(hotReloadCSVBase), 0o600)) + require.NoError(t, serverValidate(t, server)) + require.EqualValues(t, 1, fc.driftOccurrences.Load()) + + // The file does not change, but the drift persists — every subsequent + // Validate (sync start or health probe) must still count it. + require.NoError(t, serverValidate(t, server)) + require.NoError(t, serverValidate(t, server)) + require.EqualValues(t, 3, fc.driftOccurrences.Load()) + + // Drift resolves (only standard types remain): counter resets so a new + // drift episode warns immediately. + require.NoError(t, os.WriteFile(path, []byte(hotReloadCSVStandardTypes), 0o600)) + require.NoError(t, serverValidate(t, server)) + require.EqualValues(t, 0, fc.driftOccurrences.Load()) +} + func TestHotReload_InvalidFileAtStartupRequiresRestart(t *testing.T) { // Pins the known limit documented on cacheHolder: hot-load cannot // recover a process that STARTED with an invalid file. ResourceSyncers() diff --git a/pkg/connector/resources.go b/pkg/connector/resources.go index 7d1655a1..2a54adb8 100644 --- a/pkg/connector/resources.go +++ b/pkg/connector/resources.go @@ -36,7 +36,7 @@ func parseOffset(tokenStr, offsetStr string) (int, error) { return parsed, nil } -func paginate[T any](items []T, gen string, tokenStr string, tokenSize int) ([]T, string, error) { +func paginate[T any](ctx context.Context, items []T, gen string, tokenStr string, tokenSize int) ([]T, string, error) { // SDK sends Size=0 during sync_full; clamp converts it to pageSize so we never return an empty page. size := clampPageSize(tokenSize) @@ -65,6 +65,9 @@ func paginate[T any](items []T, gen string, tokenStr string, tokenSize int) ([]T } if gen == "" { offset = parsed + } else { + ctxzap.Extract(ctx).Warn( + "baton-file: legacy page token from a pre-upgrade binary; restarting listing from the beginning") } case tokenGen != gen: // The token was minted against a different cache generation: the @@ -73,7 +76,14 @@ func paginate[T any](items []T, gen string, tokenStr string, tokenSize int) ([]T // resumed after a restart). Offsets are meaningless across // generations — resuming would silently skip or duplicate items. // Restart the listing instead: store writes are idempotent - // upserts, so re-emitting earlier pages is safe. + // upserts, so re-emitting earlier pages is safe. Warn so + // operators can SEE cross-generation restarts (and the + // cross-phase consistency window documented on cacheHolder) + // instead of inferring them. + ctxzap.Extract(ctx).Warn( + "baton-file: page token was minted against a different cache generation (file changed mid-listing); restarting listing from the beginning", + zap.String("token_generation", tokenGen), + zap.String("current_generation", gen)) offset = 0 default: parsed, err := parseOffset(tokenStr, offsetStr) @@ -129,42 +139,42 @@ func (b *resourceBuilder) ResourceType(ctx context.Context) *v2.ResourceType { return b.resourceType } -func (b *resourceBuilder) List(_ context.Context, parentResourceID *v2.ResourceId, +func (b *resourceBuilder) List(ctx context.Context, parentResourceID *v2.ResourceId, opts rs.SyncOpAttrs) ([]*v2.Resource, *rs.SyncOpResults, error) { cache := b.cache.load() if cache == nil { return nil, nil, fmt.Errorf("baton-file: sync cache not initialized") } resources := cache.listIndex[listKey(b.resourceType.GetId(), parentResourceID)] - page, next, err := paginate(resources, cache.gen, opts.PageToken.Token, opts.PageToken.Size) + page, next, err := paginate(ctx, resources, cache.gen, opts.PageToken.Token, opts.PageToken.Size) if err != nil { return nil, nil, err } return page, &rs.SyncOpResults{NextPageToken: next}, nil } -func (b *resourceBuilder) Entitlements(_ context.Context, resource *v2.Resource, +func (b *resourceBuilder) Entitlements(ctx context.Context, resource *v2.Resource, opts rs.SyncOpAttrs) ([]*v2.Entitlement, *rs.SyncOpResults, error) { cache := b.cache.load() if cache == nil { return nil, nil, fmt.Errorf("baton-file: sync cache not initialized") } ents := cache.entIndex[resource.GetId().GetResource()] - page, next, err := paginate(ents, cache.gen, opts.PageToken.Token, opts.PageToken.Size) + page, next, err := paginate(ctx, ents, cache.gen, opts.PageToken.Token, opts.PageToken.Size) if err != nil { return nil, nil, err } return page, &rs.SyncOpResults{NextPageToken: next}, nil } -func (b *resourceBuilder) Grants(_ context.Context, resource *v2.Resource, +func (b *resourceBuilder) Grants(ctx context.Context, resource *v2.Resource, opts rs.SyncOpAttrs) ([]*v2.Grant, *rs.SyncOpResults, error) { cache := b.cache.load() if cache == nil { return nil, nil, fmt.Errorf("baton-file: sync cache not initialized") } grants := cache.grantsIndex[resource.GetId().GetResource()] - page, next, err := paginate(grants, cache.gen, opts.PageToken.Token, opts.PageToken.Size) + page, next, err := paginate(ctx, grants, cache.gen, opts.PageToken.Token, opts.PageToken.Size) if err != nil { return nil, nil, err } diff --git a/pkg/connector/resources_test.go b/pkg/connector/resources_test.go index d3045c86..98a3057f 100644 --- a/pkg/connector/resources_test.go +++ b/pkg/connector/resources_test.go @@ -105,7 +105,7 @@ func TestPaginate_Exhaustion(t *testing.T) { token := "" pages := 0 for { - page, next, err := paginate(items, "", token, 0) + page, next, err := paginate(context.Background(), items, "", token, 0) require.NoError(t, err) require.LessOrEqual(t, len(page), pageSize) pages++ @@ -125,7 +125,7 @@ func TestPaginate_Exhaustion(t *testing.T) { func TestPaginate_StaleOffset(t *testing.T) { items := make([]int, 50) - page, next, err := paginate(items, "", "200", 0) + page, next, err := paginate(context.Background(), items, "", "200", 0) require.NoError(t, err) require.Empty(t, page) require.Empty(t, next) @@ -134,7 +134,7 @@ func TestPaginate_StaleOffset(t *testing.T) { func TestPaginate_InvalidToken(t *testing.T) { items := []int{1, 2, 3} for _, tok := range []string{"not-a-number", "abc", "-1", "0", "gen1:abc", "gen1:-1", "gen1:0"} { - _, _, err := paginate(items, "gen1", tok, 0) + _, _, err := paginate(context.Background(), items, "gen1", tok, 0) require.Error(t, err, "token %q should return error", tok) } } @@ -143,12 +143,12 @@ func TestPaginate_GenerationStampedTokens(t *testing.T) { items := []int{10, 20, 30} // Tokens minted under a gen carry it; same-gen resume honors the offset. - page, next, err := paginate(items, "gen1", "", 2) + page, next, err := paginate(context.Background(), items, "gen1", "", 2) require.NoError(t, err) require.Equal(t, []int{10, 20}, page) require.Equal(t, "gen1:2", next) - page, next, err = paginate(items, "gen1", next, 2) + page, next, err = paginate(context.Background(), items, "gen1", next, 2) require.NoError(t, err) require.Equal(t, []int{30}, page) require.Empty(t, next) @@ -159,7 +159,7 @@ func TestPaginate_GenerationMismatchRestartsListing(t *testing.T) { // listing at the beginning instead of replaying its offset — offsets are // meaningless across generations (hot-load swapped the cache mid-listing). items := []int{10, 20, 30} - page, next, err := paginate(items, "gen2", "gen1:2", 2) + page, next, err := paginate(context.Background(), items, "gen2", "gen1:2", 2) require.NoError(t, err) require.Equal(t, []int{10, 20}, page) require.Equal(t, "gen2:2", next) @@ -171,7 +171,7 @@ func TestPaginate_LegacyNumericTokenRestartsListing(t *testing.T) { // unknowable and they only appear after a restart, when the file may // have changed. Malformed ones still error (see TestPaginate_InvalidToken). items := []int{10, 20, 30} - page, next, err := paginate(items, "gen1", "2", 2) + page, next, err := paginate(context.Background(), items, "gen1", "2", 2) require.NoError(t, err) require.Equal(t, []int{10, 20}, page) require.Equal(t, "gen1:2", next) @@ -179,7 +179,7 @@ func TestPaginate_LegacyNumericTokenRestartsListing(t *testing.T) { // Against a generation-less cache (tests build these directly), bare // numeric tokens are the cache's own mint format and must be honored — // restarting would loop forever. - page, next, err = paginate(items, "", "2", 2) + page, next, err = paginate(context.Background(), items, "", "2", 2) require.NoError(t, err) require.Equal(t, []int{30}, page) require.Empty(t, next) From 3c6ab9a5142e551908b574aa8bb306483063bc34 Mon Sep 17 00:00:00 2001 From: Matthew Scobell Date: Fri, 14 Aug 2026 10:10:35 -0400 Subject: [PATCH 09/11] fix: bound cross-generation listing restarts; test the observability MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Page tokens now carry a restart count (":[:]"): a file rewritten faster than a listing can finish previously restarted that listing forever with no forward progress; after maxListingRestarts (3) the listing fails with a clear error naming the churning file, turning unbounded work into a retryable sync failure. The count is stateless — it rides the token — so it is scoped to one listing chain and needs no shared state. - Assert the actual log emissions, not just internal state: a zap observer (go.uber.org/zap/zaptest/observer, vendored; same module version, go.mod untouched) now pins the drift Warn's logarithmic sampling (fires at occurrences 1, 2 and 4, not 3, with total_occurrences) and the two paginate restart Warns with their generation fields. Co-Authored-By: Claude Fable 5 --- pkg/connector/hot_reload_test.go | 49 +++-- pkg/connector/resources.go | 76 +++++-- pkg/connector/resources_test.go | 60 +++++- .../zap/zaptest/observer/logged_entry.go | 39 ++++ .../zap/zaptest/observer/observer.go | 203 ++++++++++++++++++ vendor/modules.txt | 1 + 6 files changed, 397 insertions(+), 31 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/connector/hot_reload_test.go b/pkg/connector/hot_reload_test.go index 38282eed..699373c8 100644 --- a/pkg/connector/hot_reload_test.go +++ b/pkg/connector/hot_reload_test.go @@ -9,7 +9,11 @@ import ( v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" "github.com/conductorone/baton-sdk/pkg/connectorbuilder" "github.com/conductorone/baton-sdk/pkg/types" + "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap/ctxzap" "github.com/stretchr/testify/require" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "go.uber.org/zap/zaptest/observer" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) @@ -272,26 +276,47 @@ func TestHotReload_SchemaDriftCheckIsLevelTriggered(t *testing.T) { // Schema drift (file types with no registered syncer) persists until // restart, so its detection must run on EVERY Validate — including ones // where the file content is unchanged — not just the sync where the - // drift first appeared. The occurrence counter drives logarithmic log - // sampling (warns at 1, 2, 4, 8, ...) and is the observable contract. + // drift first appeared. The Warn is sampled logarithmically (occurrences + // 1, 2, 4, 8, ...) so probed deployments are not flooded; the emitted + // logs are the observable contract, asserted via a zap observer. + core, logs := observer.New(zapcore.WarnLevel) + ctx := ctxzap.ToContext(context.Background(), zap.New(core)) path := filepath.Join(t.TempDir(), "input.csv") server, fc := startServer(t, path, hotReloadCSVHeaderOnly) + validate := func() { + t.Helper() + _, err := server.Validate(ctx, v2.ConnectorServiceValidateRequest_builder{}.Build()) + require.NoError(t, err) + } + driftWarns := func() []observer.LoggedEntry { + return logs.FilterMessageSnippet("not registered at startup").All() + } + // Introduce a custom "team" type that the empty start did not register. require.NoError(t, os.WriteFile(path, []byte(hotReloadCSVBase), 0o600)) - require.NoError(t, serverValidate(t, server)) - require.EqualValues(t, 1, fc.driftOccurrences.Load()) - // The file does not change, but the drift persists — every subsequent - // Validate (sync start or health probe) must still count it. - require.NoError(t, serverValidate(t, server)) - require.NoError(t, serverValidate(t, server)) - require.EqualValues(t, 3, fc.driftOccurrences.Load()) + // The file changes only before occurrence 1; the drift persists through + // the unchanged Validates that follow. Warns at 1, 2 and 4, not at 3. + validate() + require.Len(t, driftWarns(), 1) + validate() + require.Len(t, driftWarns(), 2) + validate() + require.Len(t, driftWarns(), 2, "occurrence 3 is sampled out") + validate() + require.Len(t, driftWarns(), 3) + require.EqualValues(t, 4, fc.driftOccurrences.Load()) + + for i, want := range []uint64{1, 2, 4} { + require.EqualValues(t, want, driftWarns()[i].ContextMap()["total_occurrences"]) + } - // Drift resolves (only standard types remain): counter resets so a new - // drift episode warns immediately. + // Drift resolves (only standard types remain): no new warn, and the + // counter resets so a new drift episode warns immediately. require.NoError(t, os.WriteFile(path, []byte(hotReloadCSVStandardTypes), 0o600)) - require.NoError(t, serverValidate(t, server)) + validate() + require.Len(t, driftWarns(), 3) require.EqualValues(t, 0, fc.driftOccurrences.Load()) } diff --git a/pkg/connector/resources.go b/pkg/connector/resources.go index 2a54adb8..83d0bb89 100644 --- a/pkg/connector/resources.go +++ b/pkg/connector/resources.go @@ -15,6 +15,15 @@ import ( const pageSize = 1000 +// maxListingRestarts bounds how many times one listing may restart because +// its token's cache generation went stale (the file changed mid-listing). +// Without a bound, a file rewritten continuously under frequent Validate +// probes would restart the listing forever and the sync would never finish; +// failing loudly after a few attempts turns unbounded work into a clear, +// retryable sync error. The count rides the page token, so it is stateless +// and scoped to one listing chain. +const maxListingRestarts = 3 + // clampPageSize returns requested if it is within (0, pageSize], otherwise // returns pageSize. Guards against zero (SDK default) and oversized requests. func clampPageSize(requested int) int { @@ -36,16 +45,45 @@ func parseOffset(tokenStr, offsetStr string) (int, error) { return parsed, nil } +// mintToken encodes the next-page token. An empty gen (caches built directly +// in tests) mints legacy bare-numeric tokens; a zero restart count keeps the +// canonical ":" form; restarted listings carry the count as a +// third segment so the restart bound survives statelessly across pages. +func mintToken(gen string, offset, restarts int) string { + if gen == "" { + return strconv.Itoa(offset) + } + if restarts == 0 { + return gen + ":" + strconv.Itoa(offset) + } + return gen + ":" + strconv.Itoa(offset) + ":" + strconv.Itoa(restarts) +} + +// parseRestarts validates the optional third token segment. Absent means 0. +func parseRestarts(tokenStr, restartsStr string) (int, error) { + if restartsStr == "" { + return 0, nil + } + parsed, err := strconv.Atoi(restartsStr) + if err != nil || parsed < 1 || parsed > maxListingRestarts { + return 0, fmt.Errorf("baton-file: invalid page token %q: bad restart count", tokenStr) + } + return parsed, nil +} + func paginate[T any](ctx context.Context, items []T, gen string, tokenStr string, tokenSize int) ([]T, string, error) { // SDK sends Size=0 during sync_full; clamp converts it to pageSize so we never return an empty page. size := clampPageSize(tokenSize) // Empty token means first page. Tokens minted by this version are - // ":" where gen fingerprints the cache the offset indexes - // into; bare numeric tokens are the legacy (pre-generation) format. + // ":[:]" where gen fingerprints the cache the + // offset indexes into; bare numeric tokens are the legacy + // (pre-generation) format. See mintToken. offset := 0 + restarts := 0 if tokenStr != "" { - tokenGen, offsetStr, found := strings.Cut(tokenStr, ":") + tokenGen, rest, found := strings.Cut(tokenStr, ":") + offsetStr, restartsStr, _ := strings.Cut(rest, ":") switch { case !found: // Bare numeric token. If this cache has a generation (every @@ -66,6 +104,7 @@ func paginate[T any](ctx context.Context, items []T, gen string, tokenStr string if gen == "" { offset = parsed } else { + restarts = 1 ctxzap.Extract(ctx).Warn( "baton-file: legacy page token from a pre-upgrade binary; restarting listing from the beginning") } @@ -79,18 +118,34 @@ func paginate[T any](ctx context.Context, items []T, gen string, tokenStr string // upserts, so re-emitting earlier pages is safe. Warn so // operators can SEE cross-generation restarts (and the // cross-phase consistency window documented on cacheHolder) - // instead of inferring them. + // instead of inferring them. A file rewritten faster than the + // listing can finish would restart forever, so restarts are + // bounded: failing loudly beats unbounded work. + prior, err := parseRestarts(tokenStr, restartsStr) + if err != nil { + return nil, "", err + } + restarts = prior + 1 + if restarts > maxListingRestarts { + return nil, "", fmt.Errorf( + "baton-file: listing restarted %d times because the input file kept changing mid-listing; the file is being rewritten faster than syncs can read it", prior) + } ctxzap.Extract(ctx).Warn( "baton-file: page token was minted against a different cache generation (file changed mid-listing); restarting listing from the beginning", zap.String("token_generation", tokenGen), - zap.String("current_generation", gen)) - offset = 0 + zap.String("current_generation", gen), + zap.Int("restarts", restarts)) default: parsed, err := parseOffset(tokenStr, offsetStr) if err != nil { return nil, "", err } + prior, err := parseRestarts(tokenStr, restartsStr) + if err != nil { + return nil, "", err + } offset = parsed + restarts = prior } } @@ -107,15 +162,10 @@ func paginate[T any](ctx context.Context, items []T, gen string, tokenStr string end = len(items) } - // Empty next token signals the SDK that there are no more pages. An empty - // gen (caches built directly in tests) mints legacy numeric tokens. + // Empty next token signals the SDK that there are no more pages. var next string if end < len(items) { - if gen == "" { - next = strconv.Itoa(end) - } else { - next = gen + ":" + strconv.Itoa(end) - } + next = mintToken(gen, end, restarts) } return items[offset:end], next, nil } diff --git a/pkg/connector/resources_test.go b/pkg/connector/resources_test.go index 98a3057f..8d73e6b5 100644 --- a/pkg/connector/resources_test.go +++ b/pkg/connector/resources_test.go @@ -9,7 +9,11 @@ import ( v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" "github.com/conductorone/baton-sdk/pkg/pagination" rs "github.com/conductorone/baton-sdk/pkg/types/resource" + "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap/ctxzap" "github.com/stretchr/testify/require" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "go.uber.org/zap/zaptest/observer" ) // testBuilder wraps a pre-built cache in a holder, mirroring what @@ -133,7 +137,11 @@ func TestPaginate_StaleOffset(t *testing.T) { func TestPaginate_InvalidToken(t *testing.T) { items := []int{1, 2, 3} - for _, tok := range []string{"not-a-number", "abc", "-1", "0", "gen1:abc", "gen1:-1", "gen1:0"} { + for _, tok := range []string{ + "not-a-number", "abc", "-1", "0", + "gen1:abc", "gen1:-1", "gen1:0", + "gen1:2:abc", "gen1:2:0", "gen1:2:99", "gen0:2:abc", + } { _, _, err := paginate(context.Background(), items, "gen1", tok, 0) require.Error(t, err, "token %q should return error", tok) } @@ -158,11 +166,45 @@ func TestPaginate_GenerationMismatchRestartsListing(t *testing.T) { // A token minted against a different cache generation must restart the // listing at the beginning instead of replaying its offset — offsets are // meaningless across generations (hot-load swapped the cache mid-listing). + // The restart must also be observable: exactly one Warn naming both + // generations, since it is the runtime signal for the cross-phase + // consistency window documented on cacheHolder. + core, logs := observer.New(zapcore.WarnLevel) + ctx := ctxzap.ToContext(context.Background(), zap.New(core)) + items := []int{10, 20, 30} - page, next, err := paginate(context.Background(), items, "gen2", "gen1:2", 2) + page, next, err := paginate(ctx, items, "gen2", "gen1:2", 2) require.NoError(t, err) require.Equal(t, []int{10, 20}, page) - require.Equal(t, "gen2:2", next) + require.Equal(t, "gen2:2:1", next, "restarted listings carry the restart count in the token") + + entries := logs.FilterMessageSnippet("different cache generation").All() + require.Len(t, entries, 1) + fields := entries[0].ContextMap() + require.Equal(t, "gen1", fields["token_generation"]) + require.Equal(t, "gen2", fields["current_generation"]) +} + +func TestPaginate_RestartBoundFailsLoudly(t *testing.T) { + // A file rewritten faster than a listing can finish would restart the + // listing forever; after maxListingRestarts the sync must fail with a + // clear error instead of doing unbounded work. The count rides the + // token, so it survives across pages statelessly. + ctx := context.Background() + items := []int{10, 20, 30} + + next := "gen1:2" + for round := 2; round <= maxListingRestarts+1; round++ { + gen := fmt.Sprintf("gen%d", round) + page, n, err := paginate(ctx, items, gen, next, 2) + require.NoError(t, err, "restart %d is within the bound", round-1) + require.Equal(t, []int{10, 20}, page) + require.Equal(t, fmt.Sprintf("%s:2:%d", gen, round-1), n) + next = n + } + + _, _, err := paginate(ctx, items, "genFinal", next, 2) + require.ErrorContains(t, err, "rewritten faster than syncs can read it") } func TestPaginate_LegacyNumericTokenRestartsListing(t *testing.T) { @@ -170,19 +212,25 @@ func TestPaginate_LegacyNumericTokenRestartsListing(t *testing.T) { // pre-generation binary) restart the listing: their generation is // unknowable and they only appear after a restart, when the file may // have changed. Malformed ones still error (see TestPaginate_InvalidToken). + // The restart emits exactly one observable Warn. + core, logs := observer.New(zapcore.WarnLevel) + ctx := ctxzap.ToContext(context.Background(), zap.New(core)) + items := []int{10, 20, 30} - page, next, err := paginate(context.Background(), items, "gen1", "2", 2) + page, next, err := paginate(ctx, items, "gen1", "2", 2) require.NoError(t, err) require.Equal(t, []int{10, 20}, page) - require.Equal(t, "gen1:2", next) + require.Equal(t, "gen1:2:1", next, "the legacy restart counts toward the restart bound") + require.Len(t, logs.FilterMessageSnippet("legacy page token").All(), 1) // Against a generation-less cache (tests build these directly), bare // numeric tokens are the cache's own mint format and must be honored — // restarting would loop forever. - page, next, err = paginate(context.Background(), items, "", "2", 2) + page, next, err = paginate(ctx, items, "", "2", 2) require.NoError(t, err) require.Equal(t, []int{30}, page) require.Empty(t, next) + require.Len(t, logs.FilterMessageSnippet("legacy page token").All(), 1, "generation-less caches do not warn") } // Grants() behavior tests. 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 00000000..ef89e25c --- /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 00000000..4f7ce0ec --- /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 7dac4ac5..6e835347 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -774,6 +774,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 # golang.org/x/crypto v0.54.0 ## explicit; go 1.25.0 golang.org/x/crypto/blowfish From b6e46740fc8dbbca0a5a76bf8dc761ec07cda0a9 Mon Sep 17 00:00:00 2001 From: Matthew Scobell Date: Fri, 14 Aug 2026 11:16:55 -0400 Subject: [PATCH 10/11] fix: reset restart budget on progress; verify fingerprint after parse MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - The restart budget now resets when a listing completes a page under the current generation: the churn episode is over, so the count no longer accumulates for the lifetime of a listing chain, where a checkpointed token could have turned one later file change into a hard failure. The bound is therefore consecutive-restarts-without- progress; docs updated to say exactly that, and README/connector.mdx now document the bounded-restart failure mode operators can hit. - Close a read-then-parse race in loadValidatedCache: the fingerprint was hashed from one read while the parser did its own second read, so an edit landing between them mislabeled the build — and a later revert to the hashed content would short-circuit onto the mislabeled cache indefinitely. The file is now re-hashed after parsing and the load retried (bounded) when the fingerprints disagree, so gen always describes the bytes the parser consumed. Adversarially reviewed (token state machine enumerated, concurrency stress-tested under -race through the real SDK server) and e2e-verified: all 12 example inputs produce output identical to main. Co-Authored-By: Claude Fable 5 --- README.md | 2 +- docs/connector.mdx | 2 +- pkg/connector/connector.go | 56 ++++++++++++++++++++++----------- pkg/connector/resources.go | 19 ++++++++--- pkg/connector/resources_test.go | 21 +++++++++++++ 5 files changed, 76 insertions(+), 24 deletions(-) diff --git a/README.md b/README.md index 80df5189..3a320cdf 100644 --- a/README.md +++ b/README.md @@ -41,7 +41,7 @@ baton-file -i data.xlsx ## Hot-Load (File Change Detection) -When baton-file runs as a long-lived service, edits to the input file are picked up automatically at the start of the next sync cycle — no restart required. This applies to the **data** sections: users, resources, entitlements, grants, and inheritance mappings. **Schema changes** — introducing a brand-new resource type, or changing an existing type's trait — still require a service restart, because the SDK registers resource types and their traits once at process startup. If the edited file fails validation, the sync fails with the validation error and the connector keeps serving the last successfully loaded data. If the file changes while a sync is in flight (for example via health-check revalidation), affected listings restart safely against the new contents instead of resuming into stale page offsets. A file that is valid but has no data rows is a legitimate empty state: the connector registers its standard resource types (user, group, role, app, secret), syncs empty, and hot-loads data rows as they are added — though rows using custom resource type IDs still require a restart, like any schema change. Hot-load does require a service that started successfully: if the input file is **invalid** when the service first starts, no resource types get registered and syncs fail until the file is fixed **and the service is restarted**. +When baton-file runs as a long-lived service, edits to the input file are picked up automatically at the start of the next sync cycle — no restart required. This applies to the **data** sections: users, resources, entitlements, grants, and inheritance mappings. **Schema changes** — introducing a brand-new resource type, or changing an existing type's trait — still require a service restart, because the SDK registers resource types and their traits once at process startup. If the edited file fails validation, the sync fails with the validation error and the connector keeps serving the last successfully loaded data. If the file changes while a sync is in flight (for example via health-check revalidation), affected listings restart safely against the new contents instead of resuming into stale page offsets; if the same listing restarts more than three times in a row without completing a page because the file keeps changing mid-sync, the sync fails with a "rewritten faster than syncs can read it" error and the next sync starts fresh. A file that is valid but has no data rows is a legitimate empty state: the connector registers its standard resource types (user, group, role, app, secret), syncs empty, and hot-loads data rows as they are added — though rows using custom resource type IDs still require a restart, like any schema change. Hot-load does require a service that started successfully: if the input file is **invalid** when the service first starts, no resource types get registered and syncs fail until the file is fixed **and the service is restarted**. **IMPORTANT (maintainers and AI agents):** this behavior is a deliberate contract, not an accident of implementation. The SDK calls `ResourceSyncers()` exactly once per process, so `Validate()` — which runs at the start of every sync — is the only per-sync hook and is where the file is re-read and the shared cache is republished (see `cacheHolder` in `pkg/connector/connector.go`). The construction-time load in `ResourceSyncers()` is the *first* load, never the only one: do not capture cache snapshots in builders, and do not assume the SDK refreshes data between syncs (it does not — that assumption caused a regression once already). `TestHotReload_DataChangesPickedUpBySync` in `pkg/connector/hot_reload_test.go` enforces this contract; if it fails after your change, the change is wrong, not the test. diff --git a/docs/connector.mdx b/docs/connector.mdx index a1ecf3b3..e703eb86 100644 --- a/docs/connector.mdx +++ b/docs/connector.mdx @@ -34,7 +34,7 @@ Resource types are defined by the input file (via each row's `trait`), so the ex - **Read-only.** The connector cannot create, modify, revoke, or deprovision access in the source system — it only reads what the file describes. - **Full sync only.** Every sync reads the entire file; there is no incremental/targeted sync. - **File-defined data.** C1 only knows what the file contains. Resource IDs must be globally unique across users and resources. -- **File edits apply on the next sync.** When the connector runs as a long-lived service, changes to the file's data — users, resources, entitlements, grants, and inheritance mappings — are picked up automatically at the start of the next sync, with no restart needed. Adding a new resource type, or changing an existing type's trait, does require restarting the connector. If an edited file fails validation, the sync fails with the validation error and the connector keeps serving the last successfully loaded data. A file with no data rows is valid: the connector syncs empty and picks up rows as they are added. If the file is invalid when the connector first starts (rather than edited mid-run), fix the file and restart the connector. +- **File edits apply on the next sync.** When the connector runs as a long-lived service, changes to the file's data — users, resources, entitlements, grants, and inheritance mappings — are picked up automatically at the start of the next sync, with no restart needed. Adding a new resource type, or changing an existing type's trait, does require restarting the connector. If an edited file fails validation, the sync fails with the validation error and the connector keeps serving the last successfully loaded data. A file with no data rows is valid: the connector syncs empty and picks up rows as they are added. If the file is rewritten repeatedly while a sync is running, the sync may fail with a "rewritten faster than syncs can read it" error — the next sync starts fresh; avoid continuous rewrites during syncs. If the file is invalid when the connector first starts (rather than edited mid-run), fix the file and restart the connector. ## Gather File connector credentials diff --git a/pkg/connector/connector.go b/pkg/connector/connector.go index e06a36a1..84eada1f 100644 --- a/pkg/connector/connector.go +++ b/pkg/connector/connector.go @@ -282,26 +282,46 @@ func (fc *FileConnector) Validate(ctx context.Context) (annotations.Annotations, // redundant work on health-check probes and keeps no-op Validate calls from // swapping the pointer under an in-flight sync. func loadValidatedCache(ctx context.Context, inputFilePath string, current *syncCache) (*syncCache, error) { - // Fingerprint the raw bytes to stamp this build's generation into page - // tokens (see paginate). Hashing raw content is format-agnostic and - // exact, and the extra read is cheap next to parsing — the loaders take - // paths, not bytes, so reusing one read would mean restructuring them. A - // file edit racing between the two reads mislabels one build, which at - // worst restarts a listing. - raw, err := os.ReadFile(inputFilePath) - if err != nil { - return nil, fmt.Errorf("baton-file: failed to read input file: %w", err) - } - sum := sha256.Sum256(raw) - gen := hex.EncodeToString(sum[:8]) + // The gen fingerprint stamped into page tokens (see paginate) must + // describe the bytes the parser actually consumed, but the loaders take + // a path rather than bytes, so the file is read twice: once to hash, + // once to parse. An edit racing between those reads would mislabel the + // build — and a later revert to the hashed content would then + // short-circuit onto the mislabeled cache indefinitely — so the file is + // hashed AGAIN after parsing and the load retried when the fingerprints + // disagree. Hashing raw content is format-agnostic and exact, and the + // extra reads are cheap next to parsing. + const maxLoadAttempts = 3 + var data *client.LoadedData + var gen string + for attempt := 1; ; attempt++ { + raw, err := os.ReadFile(inputFilePath) + if err != nil { + return nil, fmt.Errorf("baton-file: failed to read input file: %w", err) + } + sum := sha256.Sum256(raw) + gen = hex.EncodeToString(sum[:8]) - if current != nil && current.gen == gen { - return current, nil - } + if current != nil && current.gen == gen { + return current, nil + } - data, err := client.LoadFileData(inputFilePath) - if err != nil { - return nil, fmt.Errorf("baton-file: input file is invalid: %w", err) + data, err = client.LoadFileData(inputFilePath) + if err != nil { + return nil, fmt.Errorf("baton-file: input file is invalid: %w", err) + } + + verify, err := os.ReadFile(inputFilePath) + if err != nil { + return nil, fmt.Errorf("baton-file: failed to read input file: %w", err) + } + if sha256.Sum256(verify) == sum { + break + } + if attempt >= maxLoadAttempts { + return nil, fmt.Errorf( + "baton-file: input file kept changing while being loaded; retry once the file is stable") + } } if err := client.ValidateUniqueIDs(data); err != nil { diff --git a/pkg/connector/resources.go b/pkg/connector/resources.go index 83d0bb89..59c5ff16 100644 --- a/pkg/connector/resources.go +++ b/pkg/connector/resources.go @@ -127,8 +127,13 @@ func paginate[T any](ctx context.Context, items []T, gen string, tokenStr string } restarts = prior + 1 if restarts > maxListingRestarts { + // Failed syncs fail out; service-mode tasks each use a fresh + // temp c1z, so the next sync starts with fresh page tokens. + // (A resumed local .c1z replays its checkpointed token — + // delete the file to reset if the error persists there.) return nil, "", fmt.Errorf( - "baton-file: listing restarted %d times because the input file kept changing mid-listing; the file is being rewritten faster than syncs can read it", prior) + "baton-file: listing restarted %d times because the input file kept changing mid-listing; "+ + "the file is being rewritten faster than syncs can read it", prior) } ctxzap.Extract(ctx).Warn( "baton-file: page token was minted against a different cache generation (file changed mid-listing); restarting listing from the beginning", @@ -140,12 +145,18 @@ func paginate[T any](ctx context.Context, items []T, gen string, tokenStr string if err != nil { return nil, "", err } - prior, err := parseRestarts(tokenStr, restartsStr) - if err != nil { + // Validate the restart segment, but do not carry it forward: + // reaching this branch means the listing completed a page under + // the current generation — the churn episode is over, so the + // restart budget resets. Without the reset, the count would + // accumulate for the lifetime of a listing chain and a + // checkpointed token could turn one later file change into a + // hard failure. + if _, err := parseRestarts(tokenStr, restartsStr); err != nil { return nil, "", err } offset = parsed - restarts = prior + restarts = 0 } } diff --git a/pkg/connector/resources_test.go b/pkg/connector/resources_test.go index 8d73e6b5..d8129ffa 100644 --- a/pkg/connector/resources_test.go +++ b/pkg/connector/resources_test.go @@ -207,6 +207,27 @@ func TestPaginate_RestartBoundFailsLoudly(t *testing.T) { require.ErrorContains(t, err, "rewritten faster than syncs can read it") } +func TestPaginate_ProgressResetsRestartBudget(t *testing.T) { + // Completing a page under the current generation means the churn episode + // is over: the restart count must reset rather than accumulate for the + // lifetime of the listing chain — otherwise a checkpointed token could + // turn one later file change into a hard failure. + ctx := context.Background() + items := []int{10, 20, 30, 40, 50} + + // Same-generation resume with a maxed budget: progress clears it. + page, next, err := paginate(ctx, items, "gen2", "gen2:2:3", 2) + require.NoError(t, err) + require.Equal(t, []int{30, 40}, page) + require.Equal(t, "gen2:4", next, "the restart count is dropped after same-generation progress") + + // A later mismatch starts counting from 1 again instead of failing. + page, next, err = paginate(ctx, items, "gen3", next, 2) + require.NoError(t, err) + require.Equal(t, []int{10, 20}, page) + require.Equal(t, "gen3:2:1", next) +} + func TestPaginate_LegacyNumericTokenRestartsListing(t *testing.T) { // Against a generation-stamped cache, bare numeric tokens (minted by a // pre-generation binary) restart the listing: their generation is From c865274877a470cc34f072cd19ebc3796e04a102 Mon Sep 17 00:00:00 2001 From: Matthew Scobell Date: Fri, 14 Aug 2026 12:04:18 -0400 Subject: [PATCH 11/11] fix: retry torn reads during load; document the trade-offs precisely MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - A rewrite landing during the parse can tear the read and make a valid file transiently unparseable — at startup that needlessly forced a process restart. On parse failure the file is now re-hashed: changed (or momentarily unreadable) bytes mean a racing write and the load retries; unchanged bytes mean the file is genuinely invalid and the error is returned as before. - Record the restart-bound liveness adjudication on maxListingRestarts: the bound is deliberately consecutive (progress resets it); decaying instead of resetting fails the same alternating-churn pattern it purports to fix, and a cumulative count reintroduces the checkpoint hard-failure. Documented so it is not "fixed" without solving that tension. - Docs precision: the restart bound applies to consecutive restarts with no page under an unchanged file in between, and the new "kept changing while being loaded" failure is now documented in README and connector.mdx. Co-Authored-By: Claude Fable 5 --- README.md | 2 +- docs/connector.mdx | 2 +- pkg/connector/connector.go | 12 ++++++++++++ pkg/connector/resources.go | 11 +++++++++++ 4 files changed, 25 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 3a320cdf..6d1f5ada 100644 --- a/README.md +++ b/README.md @@ -41,7 +41,7 @@ baton-file -i data.xlsx ## Hot-Load (File Change Detection) -When baton-file runs as a long-lived service, edits to the input file are picked up automatically at the start of the next sync cycle — no restart required. This applies to the **data** sections: users, resources, entitlements, grants, and inheritance mappings. **Schema changes** — introducing a brand-new resource type, or changing an existing type's trait — still require a service restart, because the SDK registers resource types and their traits once at process startup. If the edited file fails validation, the sync fails with the validation error and the connector keeps serving the last successfully loaded data. If the file changes while a sync is in flight (for example via health-check revalidation), affected listings restart safely against the new contents instead of resuming into stale page offsets; if the same listing restarts more than three times in a row without completing a page because the file keeps changing mid-sync, the sync fails with a "rewritten faster than syncs can read it" error and the next sync starts fresh. A file that is valid but has no data rows is a legitimate empty state: the connector registers its standard resource types (user, group, role, app, secret), syncs empty, and hot-loads data rows as they are added — though rows using custom resource type IDs still require a restart, like any schema change. Hot-load does require a service that started successfully: if the input file is **invalid** when the service first starts, no resource types get registered and syncs fail until the file is fixed **and the service is restarted**. +When baton-file runs as a long-lived service, edits to the input file are picked up automatically at the start of the next sync cycle — no restart required. This applies to the **data** sections: users, resources, entitlements, grants, and inheritance mappings. **Schema changes** — introducing a brand-new resource type, or changing an existing type's trait — still require a service restart, because the SDK registers resource types and their traits once at process startup. If the edited file fails validation, the sync fails with the validation error and the connector keeps serving the last successfully loaded data. If the file changes while a sync is in flight (for example via health-check revalidation), affected listings restart safely against the new contents instead of resuming into stale page offsets; if the same listing restarts more than three times in a row — with no page served under an unchanged file in between — the sync fails with a "rewritten faster than syncs can read it" error and the next sync starts fresh. A file that cannot even be read consistently (it keeps changing during the load itself) fails the sync with an "input file kept changing while being loaded" error, and likewise recovers on the next sync once the file is stable. A file that is valid but has no data rows is a legitimate empty state: the connector registers its standard resource types (user, group, role, app, secret), syncs empty, and hot-loads data rows as they are added — though rows using custom resource type IDs still require a restart, like any schema change. Hot-load does require a service that started successfully: if the input file is **invalid** when the service first starts, no resource types get registered and syncs fail until the file is fixed **and the service is restarted**. **IMPORTANT (maintainers and AI agents):** this behavior is a deliberate contract, not an accident of implementation. The SDK calls `ResourceSyncers()` exactly once per process, so `Validate()` — which runs at the start of every sync — is the only per-sync hook and is where the file is re-read and the shared cache is republished (see `cacheHolder` in `pkg/connector/connector.go`). The construction-time load in `ResourceSyncers()` is the *first* load, never the only one: do not capture cache snapshots in builders, and do not assume the SDK refreshes data between syncs (it does not — that assumption caused a regression once already). `TestHotReload_DataChangesPickedUpBySync` in `pkg/connector/hot_reload_test.go` enforces this contract; if it fails after your change, the change is wrong, not the test. diff --git a/docs/connector.mdx b/docs/connector.mdx index e703eb86..9c9a30c0 100644 --- a/docs/connector.mdx +++ b/docs/connector.mdx @@ -34,7 +34,7 @@ Resource types are defined by the input file (via each row's `trait`), so the ex - **Read-only.** The connector cannot create, modify, revoke, or deprovision access in the source system — it only reads what the file describes. - **Full sync only.** Every sync reads the entire file; there is no incremental/targeted sync. - **File-defined data.** C1 only knows what the file contains. Resource IDs must be globally unique across users and resources. -- **File edits apply on the next sync.** When the connector runs as a long-lived service, changes to the file's data — users, resources, entitlements, grants, and inheritance mappings — are picked up automatically at the start of the next sync, with no restart needed. Adding a new resource type, or changing an existing type's trait, does require restarting the connector. If an edited file fails validation, the sync fails with the validation error and the connector keeps serving the last successfully loaded data. A file with no data rows is valid: the connector syncs empty and picks up rows as they are added. If the file is rewritten repeatedly while a sync is running, the sync may fail with a "rewritten faster than syncs can read it" error — the next sync starts fresh; avoid continuous rewrites during syncs. If the file is invalid when the connector first starts (rather than edited mid-run), fix the file and restart the connector. +- **File edits apply on the next sync.** When the connector runs as a long-lived service, changes to the file's data — users, resources, entitlements, grants, and inheritance mappings — are picked up automatically at the start of the next sync, with no restart needed. Adding a new resource type, or changing an existing type's trait, does require restarting the connector. If an edited file fails validation, the sync fails with the validation error and the connector keeps serving the last successfully loaded data. A file with no data rows is valid: the connector syncs empty and picks up rows as they are added. If the file is rewritten repeatedly while a sync is running, the sync may fail with a "rewritten faster than syncs can read it" or "kept changing while being loaded" error — the next sync starts fresh; avoid continuous rewrites during syncs. If the file is invalid when the connector first starts (rather than edited mid-run), fix the file and restart the connector. ## Gather File connector credentials diff --git a/pkg/connector/connector.go b/pkg/connector/connector.go index 84eada1f..6e70f40a 100644 --- a/pkg/connector/connector.go +++ b/pkg/connector/connector.go @@ -308,6 +308,18 @@ func loadValidatedCache(ctx context.Context, inputFilePath string, current *sync data, err = client.LoadFileData(inputFilePath) if err != nil { + // A rewrite landing during the parse can tear the read and make + // a perfectly valid file transiently unparseable — at startup + // that would needlessly force a process restart. Distinguish a + // racing write from a genuinely invalid file by re-hashing: + // changed (or momentarily unreadable) bytes mean a racing write, + // so retry; unchanged bytes mean the parse error is real. + if attempt < maxLoadAttempts { + verify, verr := os.ReadFile(inputFilePath) + if verr != nil || sha256.Sum256(verify) != sum { + continue + } + } return nil, fmt.Errorf("baton-file: input file is invalid: %w", err) } diff --git a/pkg/connector/resources.go b/pkg/connector/resources.go index 59c5ff16..dcfa5d63 100644 --- a/pkg/connector/resources.go +++ b/pkg/connector/resources.go @@ -22,6 +22,17 @@ const pageSize = 1000 // failing loudly after a few attempts turns unbounded work into a clear, // retryable sync error. The count rides the page token, so it is stateless // and scoped to one listing chain. +// +// The bound counts CONSECUTIVE restarts: a page served under an unchanged +// generation resets it (see paginate), so a checkpointed token cannot turn +// one later file change into a hard failure. This is a deliberate +// liveness-over-termination choice — churn timed to alternate with single +// pages of progress can in principle defeat the bound, but that rhythm is +// implausible against microsecond in-memory pages, task-level timeouts bound +// a hung sync externally, and decaying the count instead of resetting it +// fails the same alternating pattern (mismatch +1 / progress -1 oscillates +// below any bound forever) while a cumulative count reintroduces the +// checkpoint hard-failure. Do not "fix" this without solving that tension. const maxListingRestarts = 3 // clampPageSize returns requested if it is within (0, pageSize], otherwise