diff --git a/README.md b/README.md index c266f8d1..6d1f5ada 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, 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. + ## 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/docs/connector.mdx b/docs/connector.mdx index c66308b0..9c9a30c0 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. 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 13524cf7..6e70f40a 100644 --- a/pkg/connector/connector.go +++ b/pkg/connector/connector.go @@ -2,10 +2,14 @@ package connector import ( "context" + "crypto/sha256" + "encoding/hex" "fmt" "os" "sort" "strings" + "sync" + "sync/atomic" "github.com/conductorone/baton-file/pkg/client" v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" @@ -24,9 +28,122 @@ import ( type FileConnector struct { inputFilePath string - validatedData *client.LoadedData + 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 + + // 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{} + + // 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 +// 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 +// 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, or a trait change to +// 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 +// 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). +// +// 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. 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 +// 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. 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] +} + +// 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 +174,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,9 +232,108 @@ 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). + cache, changed, err := fc.refresh(ctx) if err != nil { - return nil, fmt.Errorf("baton-file: input file is invalid: %w", err) + return nil, err + } + 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", + 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. 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 +} + +// loadValidatedCache reads the input file, runs all cross-record validations, +// 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) { + // 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 + } + + 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) + } + + 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 { @@ -134,61 +356,80 @@ 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 = gen + 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 { - var err error - loadedData, err = client.LoadFileData(fc.inputFilePath) - if err != nil { - l.Error("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) + // 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 { - l.Error("baton-file: failed to build sync cache", zap.Error(err)) + // 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 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 - for _, rt := range cache.resourceTypes { - syncers = append(syncers, &resourceBuilder{cache: cache, resourceType: rt}) + 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))) } - 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/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..699373c8 --- /dev/null +++ b/pkg/connector/hot_reload_test.go @@ -0,0 +1,343 @@ +package connector + +import ( + "context" + "os" + "path/filepath" + "testing" + + 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" +) + +// Hot-load contract tests (see cacheHolder in connector.go). +// +// 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 → 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,,,,,, +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,,,,,,,, +` + +// 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) { + t.Helper() + require.NoError(t, os.WriteFile(path, []byte(csv), 0o600)) + fc := &FileConnector{inputFilePath: path} + 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 + } + } +} + +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 + } + } +} + +func findResource(t *testing.T, server types.ConnectorServer, typeID, id string) *v2.Resource { + t.Helper() + for _, r := range listAll(t, server, typeID, 0) { + if r.GetId().GetResource() == id { + return r + } + } + t.Fatalf("resource %s/%s not found", typeID, id) + return nil +} + +func TestHotReload_DataChangesPickedUpBySync(t *testing.T) { + path := filepath.Join(t.TempDir(), "input.csv") + server, _ := startServer(t, path, hotReloadCSVBase) + + 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 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, 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") + server, _ := startServer(t, path, hotReloadCSVBase) + + // Fetch the first page (1 of 2 users) so a token is in flight. + first, err := server.ListResources(ctx, v2.ResourcesServiceListResourcesRequest_builder{ + ResourceTypeId: "user", + PageSize: 1, + }.Build()) + require.NoError(t, err) + 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. + require.NoError(t, os.WriteFile(path, []byte(hotReloadCSVUpdated), 0o600)) + 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. + seen := map[string]int{} + token := first.GetNextPageToken() + for { + resp, err := server.ListResources(ctx, v2.ResourcesServiceListResourcesRequest_builder{ + ResourceTypeId: "user", + PageToken: token, + PageSize: 1, + }.Build()) + require.NoError(t, err) + for _, r := range resp.GetList() { + seen[r.GetId().GetResource()]++ + } + if token = resp.GetNextPageToken(); token == "" { + break + } + } + require.Equal(t, map[string]int{"alex.taylor": 1, "sam.johnson": 1, "jordan.lee": 1}, seen) +} + +func TestHotReload_UnchangedFileSkipsRebuildAndSwap(t *testing.T) { + path := filepath.Join(t.TempDir(), "input.csv") + server, fc := startServer(t, path, hotReloadCSVBase) + + // A Validate against unchanged content (every health probe, and every + // 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() + 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)) + require.NoError(t, serverValidate(t, server)) + require.NotSame(t, before, fc.cache.load()) +} + +func TestHotReload_InvalidFileKeepsLastGoodCache(t *testing.T) { + path := filepath.Join(t.TempDir(), "input.csv") + server, _ := startServer(t, path, hotReloadCSVBase) + + 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_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_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 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)) + + // 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): no new warn, and the + // counter resets so a new drift episode warns immediately. + require.NoError(t, os.WriteFile(path, []byte(hotReloadCSVStandardTypes), 0o600)) + validate() + require.Len(t, driftWarns(), 3) + 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() + // 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)) + + // 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)) + + _, err = server.ListResourceTypes(ctx, v2.ResourceTypesServiceListResourceTypesRequest_builder{}.Build()) + require.Equal(t, codes.FailedPrecondition, status.Code(err)) +} diff --git a/pkg/connector/resources.go b/pkg/connector/resources.go index d3f564a0..dcfa5d63 100644 --- a/pkg/connector/resources.go +++ b/pkg/connector/resources.go @@ -15,6 +15,26 @@ 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. +// +// 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 // returns pageSize. Guards against zero (SDK default) and oversized requests. func clampPageSize(requested int) int { @@ -24,21 +44,131 @@ 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 +} + +// 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; 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. See mintToken. offset := 0 + restarts := 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, rest, found := strings.Cut(tokenStr, ":") + offsetStr, restartsStr, _ := strings.Cut(rest, ":") + switch { + case !found: + // 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 + } + 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") + } + 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. Warn so + // operators can SEE cross-generation restarts (and the + // cross-phase consistency window documented on cacheHolder) + // 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 { + // 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) + } + 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), + zap.Int("restarts", restarts)) + default: + parsed, err := parseOffset(tokenStr, offsetStr) + if err != nil { + return nil, "", err + } + // 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 = 0 } - 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 @@ -57,13 +187,23 @@ func paginate[T any](items []T, tokenStr string, tokenSize int) ([]T, string, er // Empty next token signals the SDK that there are no more pages. var next string if end < len(items) { - next = strconv.Itoa(end) + next = mintToken(gen, end, restarts) } 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). + // + // 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 } @@ -71,36 +211,56 @@ 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) { - 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(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) { - 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(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) { - 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(ctx, grants, cache.gen, opts.PageToken.Token, opts.PageToken.Size) if err != nil { return nil, nil, err } 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) @@ -110,7 +270,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 +335,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 +372,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 +380,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 +388,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 +403,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..d8129ffa 100644 --- a/pkg/connector/resources_test.go +++ b/pkg/connector/resources_test.go @@ -9,9 +9,21 @@ 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 +// 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 +109,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++ @@ -117,7 +129,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) @@ -125,12 +137,123 @@ 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", + "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) } } +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(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(context.Background(), 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). + // 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(ctx, items, "gen2", "gen1:2", 2) + require.NoError(t, err) + require.Equal(t, []int{10, 20}, page) + 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_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 + // 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(ctx, items, "gen1", "2", 2) + require.NoError(t, err) + require.Equal(t, []int{10, 20}, page) + 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(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. func TestGrants_InheritanceMapping_ExpandableAnnotation(t *testing.T) { @@ -157,7 +280,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 +320,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 +344,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 +374,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 +401,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 +426,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 +450,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 +471,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) 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