c1z replay

.gitignore

@@ -9,6 +9,8 @@
 *.dylib
 build/
 *.c1z
+# Frozen test fixtures are the exception (version-skew corpus etc.).
+!pkg/sync/testdata/*.c1z
 *.db
 *.csv
 *.xlsx

cmd/baton/main.go

@@ -34,6 +34,7 @@ func main() {
 	cliCmd.AddCommand(accessCmd())
 	cliCmd.AddCommand(dumpDBCmd())
 	cliCmd.AddCommand(syncsCmd())
+	cliCmd.AddCommand(sourceCacheCmd())
 	cliCmd.AddCommand(optimizeDb())
 	cliCmd.AddCommand(toPebbleCmd())
 	cliCmd.AddCommand(explorerCmd())

cmd/baton/source_cache.go

@@ -0,0 +1,255 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"sort"
+	"strings"
+	"text/tabwriter"
+	"time"
+
+	"github.com/spf13/cobra"
+
+	"github.com/conductorone/baton-sdk/pkg/dotc1z"
+	"github.com/conductorone/baton-sdk/pkg/logging"
+	"github.com/conductorone/baton-sdk/pkg/sourcecache"
+	sdkSync "github.com/conductorone/baton-sdk/pkg/sync"
+)
+
+// source-cache: the attribution surface for source-cache replay. Rows in
+// a c1z are stamped with the scope that produced them and the manifest
+// records each scope's validator + write time — this command joins the
+// two so "why is this row missing / stale?" becomes a per-scope answer:
+// when was the scope last actually fetched, what validator did it record,
+// how many rows does it hold, and is its index/manifest state consistent.
+//
+// Exits non-zero when any orphan-index scope is found (ingestion
+// invariant I6) on a file that is ELIGIBLE as a replay source: such a
+// file would poison a future sync's replay. Files the metadata gate
+// already refuses (compaction folds — which legitimately clear the
+// manifest while keeping scope stamps — and partial/derived syncs) can
+// never be replayed from, so their index/manifest divergence is reported
+// as "stale-index" without failing the audit.
+func sourceCacheCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "source-cache",
+		Short: "Audit the C1Z's source-cache replay state (scopes, validators, row counts, consistency)",
+		RunE:  runSourceCacheAudit,
+	}
+	return cmd
+}
+
+type sourceCacheScopeReport struct {
+	RowKind      string    `json:"row_kind"`
+	ScopeKey     string    `json:"scope_key"`
+	Rows         int       `json:"rows"`
+	Validator    string    `json:"cache_validator,omitempty"`
+	DiscoveredAt time.Time `json:"discovered_at,omitempty"`
+	// Status: "ok" (manifest + rows), "zero-row" (manifest entry with no
+	// stamped rows — legal: the scope legitimately returned nothing),
+	// "invalidated" (the dangling-reference drops removed rows from this
+	// scope, so its entry is marked to miss and re-fetch cold next sync —
+	// legal, informational), "orphan-index" (stamped rows with NO
+	// manifest entry on a replay-eligible file — an invariant violation
+	// that would poison the next sync's replay; see ingestion invariant
+	// I6), or "stale-index" (the same shape on a file the replay metadata
+	// gate already refuses, e.g. a compaction fold whose manifest was
+	// deliberately cleared — informational, never a failure).
+	Status string `json:"status"`
+}
+
+type sourceCacheAuditReport struct {
+	SyncID string `json:"sync_id,omitempty"`
+	// SyncType / Compacted / UsableAsReplaySource describe whether a
+	// FUTURE sync may use this file as its replay source (the metadata
+	// gate; see c1zstore.SyncRun.UsableAsReplaySource, plus the
+	// compaction-provenance token check the syncer applies for artifacts
+	// produced by pre-compacted-flag compactors).
+	SyncType             string                   `json:"sync_type,omitempty"`
+	Compacted            bool                     `json:"compacted"`
+	UsableAsReplaySource bool                     `json:"usable_as_replay_source"`
+	ReplayUnusableReason string                   `json:"replay_unusable_reason,omitempty"`
+	Scopes               []sourceCacheScopeReport `json:"scopes"`
+	OrphanIndexScopes    int                      `json:"orphan_index_scopes"`
+	StaleIndexScopes     int                      `json:"stale_index_scopes,omitempty"`
+}
+
+func runSourceCacheAudit(cmd *cobra.Command, args []string) error {
+	ctx, err := logging.Init(context.Background(), logging.WithLogFormat("console"), logging.WithLogLevel("error"))
+	if err != nil {
+		return err
+	}
+	c1zPath, err := cmd.Flags().GetString("file")
+	if err != nil {
+		return err
+	}
+	outputFormat, err := cmd.Flags().GetString("output-format")
+	if err != nil {
+		return err
+	}
+
+	store, err := openReadOnlyC1ZStore(ctx, c1zPath)
+	if err != nil {
+		return err
+	}
+	defer store.Close(ctx)
+
+	inspector, ok := store.(dotc1z.SourceCacheInspector)
+	if !ok {
+		return fmt.Errorf("source-cache state exists only in pebble-format c1z files; %q is not one", c1zPath)
+	}
+
+	report := sourceCacheAuditReport{}
+	// The newest FINISHED sync of any type: it is the sync the store
+	// opened (see openReadOnlyC1ZStore's SetCurrentSync) and the one the
+	// replay metadata gate would evaluate — LatestFullSync could describe
+	// an older sync than the state being inspected, or nothing at all on
+	// partial/derived artifacts.
+	//
+	// A metadata READ FAILURE is fatal: replay eligibility gates whether
+	// orphan indexes are a hard failure or informational, so silently
+	// defaulting eligibility to false would let the audit exit zero on a
+	// file it could not actually judge.
+	latest, err := store.SyncMeta().LatestFinishedSyncOfAnyType(ctx)
+	if err != nil {
+		return fmt.Errorf("reading sync run metadata (needed to judge replay eligibility): %w", err)
+	}
+	if latest != nil {
+		report.SyncID = latest.ID
+		report.SyncType = string(latest.Type)
+		report.Compacted = latest.Compacted
+		// The syncer's own metadata gate (type, compacted flag,
+		// compaction-provenance token) — shared, so audit and syncer can
+		// never drift.
+		report.ReplayUnusableReason = sdkSync.ReplaySourceRunUnusableReason(latest)
+		report.UsableAsReplaySource = report.ReplayUnusableReason == ""
+	}
+	if report.UsableAsReplaySource {
+		// The replay-compatibility gate's offline-decidable half: key
+		// PRESENCE. An exact-match verdict depends on the NEXT run's
+		// connector declarations (cache generation, config fingerprint,
+		// selection), which no offline audit can know — but an ABSENT
+		// key can never match any future key, so the syncer will always
+		// run cold from this file (pre-compat-key SDKs, compaction
+		// folds). Reporting such a file usable=true would be wrong.
+		if _, compatFound, compatErr := inspector.GetSourceCacheCompat(ctx); compatErr != nil {
+			return fmt.Errorf("reading replay-compatibility key: %w", compatErr)
+		} else if !compatFound {
+			report.UsableAsReplaySource = false
+			report.ReplayUnusableReason = "no replay-compatibility key recorded (pre-key SDK or compaction fold); the syncer treats an absent key as a mismatch and always runs cold"
+		}
+	}
+
+	manifest, err := inspector.SourceCacheManifestSnapshot(ctx)
+	if err != nil {
+		return fmt.Errorf("reading source-cache manifest: %w", err)
+	}
+	indexCounts, err := inspector.SourceScopeIndexSnapshot(ctx)
+	if err != nil {
+		return fmt.Errorf("reading scope indexes: %w", err)
+	}
+	orphans, err := inspector.SourceCacheOrphanScopes(ctx)
+	if err != nil {
+		return fmt.Errorf("checking scope consistency: %w", err)
+	}
+
+	for key, snapEntry := range manifest {
+		kind, scope, found := strings.Cut(key, "\x00")
+		if !found {
+			return fmt.Errorf("malformed manifest snapshot key %q", key)
+		}
+		rep := sourceCacheScopeReport{RowKind: kind, ScopeKey: scope, Rows: indexCounts[kind][scope]}
+		entry, entryFound, err := inspector.LookupSourceCacheEntry(ctx, sourcecache.RowKind(kind), scope)
+		if err != nil {
+			return fmt.Errorf("reading manifest entry for %s/%q: %w", kind, scope, err)
+		}
+		if entryFound {
+			rep.Validator = entry.CacheValidator
+			rep.DiscoveredAt = entry.DiscoveredAt
+		}
+		switch {
+		case snapEntry.Invalidated:
+			// The lookup surface reports invalidated entries as misses;
+			// surface the retained (stale) validator for forensics.
+			rep.Validator = snapEntry.CacheValidator
+			rep.Status = "invalidated"
+		case rep.Rows == 0:
+			rep.Status = "zero-row"
+		default:
+			rep.Status = "ok"
+		}
+		report.Scopes = append(report.Scopes, rep)
+	}
+	for kind, scopes := range orphans {
+		for _, scope := range scopes {
+			status := "orphan-index"
+			if !report.UsableAsReplaySource {
+				// Replay from this file is already impossible (compaction
+				// fold, partial/derived sync): the divergence cannot poison
+				// anything. Compaction folds in particular ALWAYS look like
+				// this — the fold clears the manifest and keeps scope
+				// stamps by design (see pebble.ClearSourceCacheEntries).
+				status = "stale-index"
+			}
+			report.Scopes = append(report.Scopes, sourceCacheScopeReport{
+				RowKind: kind, ScopeKey: scope, Rows: indexCounts[kind][scope], Status: status,
+			})
+			if status == "orphan-index" {
+				report.OrphanIndexScopes++
+			} else {
+				report.StaleIndexScopes++
+			}
+		}
+	}
+	sort.Slice(report.Scopes, func(i, j int) bool {
+		if report.Scopes[i].RowKind != report.Scopes[j].RowKind {
+			return report.Scopes[i].RowKind < report.Scopes[j].RowKind
+		}
+		return report.Scopes[i].ScopeKey < report.Scopes[j].ScopeKey
+	})
+
+	if outputFormat == "json" {
+		enc := json.NewEncoder(os.Stdout)
+		enc.SetIndent("", "  ")
+		if err := enc.Encode(report); err != nil {
+			return err
+		}
+	} else if err := printSourceCacheAudit(report); err != nil {
+		return err
+	}
+	if report.OrphanIndexScopes > 0 {
+		return fmt.Errorf("%d orphan-index scope(s) found (ingestion invariant I6); a future sync replaying from this file could silently drop or resurrect rows", report.OrphanIndexScopes)
+	}
+	return nil
+}
+
+func printSourceCacheAudit(report sourceCacheAuditReport) error {
+	if report.SyncID != "" {
+		fmt.Fprintf(os.Stdout, "sync %s type=%s compacted=%v usable-as-replay-source=%v\n",
+			report.SyncID, report.SyncType, report.Compacted, report.UsableAsReplaySource)
+		if report.ReplayUnusableReason != "" {
+			fmt.Fprintf(os.Stdout, "  (not usable: %s)\n", report.ReplayUnusableReason)
+		}
+		fmt.Fprintln(os.Stdout)
+	}
+	if len(report.Scopes) == 0 {
+		fmt.Fprintln(os.Stdout, "no source-cache state (cold artifact: every scope will fetch fresh next sync)")
+		return nil
+	}
+	w := tabwriter.NewWriter(os.Stdout, 2, 4, 2, ' ', 0)
+	fmt.Fprintln(w, "KIND\tSCOPE\tROWS\tSTATUS\tDISCOVERED\tVALIDATOR")
+	for _, r := range report.Scopes {
+		validator := r.Validator
+		if len(validator) > 48 {
+			validator = validator[:45] + "..."
+		}
+		discovered := ""
+		if !r.DiscoveredAt.IsZero() {
+			discovered = r.DiscoveredAt.UTC().Format(time.RFC3339)
+		}
+		fmt.Fprintf(w, "%s\t%s\t%d\t%s\t%s\t%s\n", r.RowKind, r.ScopeKey, r.Rows, r.Status, discovered, validator)
+	}
+	return w.Flush()
+}

internal/connector/connector.go

@@ -23,11 +23,13 @@ import (
 
 	connectorV2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2"
 	connectorwrapperV1 "github.com/conductorone/baton-sdk/pb/c1/connector_wrapper/v1"
+	batonV1 "github.com/conductorone/baton-sdk/pb/c1/connectorapi/baton/v1"
 	ratelimitV1 "github.com/conductorone/baton-sdk/pb/c1/ratelimit/v1"
 	tlsV1 "github.com/conductorone/baton-sdk/pb/c1/utls/v1"
 	"github.com/conductorone/baton-sdk/pkg/bid"
 	ratelimit2 "github.com/conductorone/baton-sdk/pkg/ratelimit"
 	"github.com/conductorone/baton-sdk/pkg/session"
+	"github.com/conductorone/baton-sdk/pkg/sourcecache"
 	"github.com/conductorone/baton-sdk/pkg/types"
 	"github.com/conductorone/baton-sdk/pkg/types/sessions"
 	"github.com/conductorone/baton-sdk/pkg/ugrpc"
@@ -54,20 +56,43 @@ type connectorClient struct {
 	connectorV2.TicketsServiceClient
 	connectorV2.ActionServiceClient
 
-	sessionStoreSetter sessions.SetSessionStore // this is the session store server
+	sessionStoreSetter sessions.SetSessionStore      // this is the session store server
+	sourceCacheSetter  sourcecache.SourceCacheSetter // this is the source-cache lookup server
 }
 
 var _ sessions.SetSessionStore = (*connectorClient)(nil)
+var _ sourcecache.SourceCacheSetter = (*connectorClient)(nil)
 var _ SetSessionStoreSetter = (*connectorClient)(nil)
+var _ SetSourceCacheSetter = (*connectorClient)(nil)
 
 type SetSessionStoreSetter interface {
 	SetSessionStoreSetter(setsessionStoreSetter sessions.SetSessionStore)
 }
 
+type SetSourceCacheSetter interface {
+	SetSourceCacheSetter(sourceCacheSetter sourcecache.SourceCacheSetter)
+}
+
 func (c *connectorClient) SetSessionStoreSetter(sessionStoreSetter sessions.SetSessionStore) {
 	c.sessionStoreSetter = sessionStoreSetter
 }
 
+func (c *connectorClient) SetSourceCacheSetter(sourceCacheSetter sourcecache.SourceCacheSetter) {
+	c.sourceCacheSetter = sourceCacheSetter
+}
+
+// SetSourceCache forwards the syncer's per-sync lookup to whatever
+// receives it: the subprocess-mode BatonSourceCacheService server, or the
+// in-process builder. A nil setter means the connector never opted in;
+// the syncer calls this unconditionally, so stay quiet at debug level.
+func (c *connectorClient) SetSourceCache(ctx context.Context, lookup sourcecache.Lookup) {
+	if c.sourceCacheSetter == nil {
+		ctxzap.Extract(ctx).Debug("connectorClient's source cache setter is nil — connector did not opt into source caching")
+		return
+	}
+	c.sourceCacheSetter.SetSourceCache(ctx, lookup)
+}
+
 func (c *connectorClient) SetSessionStore(ctx context.Context, store sessions.SessionStore) {
 	if c.sessionStoreSetter == nil {
 		// Demoted from Warn to Debug: this path is the normal case for
@@ -109,7 +134,8 @@ type wrapper struct {
 
 	now func() time.Time
 
-	SessionServer sessions.SetSessionStore
+	SessionServer     sessions.SetSessionStore
+	SourceCacheServer sourcecache.SourceCacheSetter
 }
 
 type Option func(ctx context.Context, w *wrapper) error
@@ -197,6 +223,12 @@ func NewWrapper(ctx context.Context, server interface{}, opts ...Option) (*wrapp
 		server: connectorServer,
 		now:    time.Now,
 	}
+	// In-process delivery: a connectorbuilder-based server implements
+	// sourcecache.SourceCacheSetter itself, so the syncer's per-sync lookup can be
+	// installed without the subprocess gRPC hop.
+	if sourceCacheServer, ok := connectorServer.(sourcecache.SourceCacheSetter); ok {
+		w.SourceCacheServer = sourceCacheServer
+	}
 
 	for _, o := range opts {
 		err := o(ctx, w)
@@ -276,18 +308,29 @@ func (cw *wrapper) runServer(ctx context.Context, serverCred *tlsV1.Credential)
 			return 0, fmt.Errorf("failed to create session listener config: %w", err)
 		}
 
-		// TODO(kans): block until we send a request or something/error handling in general.
+		// One listener serves BatonSessionService (connector session data)
+		// and BatonSourceCacheService (source-cache scope lookups). Keeping
+		// them as separate RPCs keeps validator lookups out of the
+		// connector's local MemorySessionCache and its TTL/eviction rules.
 		l.Info("starting session store server")
-		server := session.NewGRPCSessionServer()
-		cw.SessionServer = server
+		sessionServer := session.NewGRPCSessionServer()
+		sourceCacheServer := sourcecache.NewGRPCServer()
+		cw.SessionServer = sessionServer
+		cw.SourceCacheServer = sourceCacheServer
 		go func() {
 			defer sessionListener.Close()
-			serverErr := session.StartGRPCSessionServerWithOptions(ctx, sessionListener, server,
+			grpcServer := grpc.NewServer(
 				grpc.Creds(credentials.NewTLS(tlsConfig)),
 				grpc.ChainUnaryInterceptor(ugrpc.UnaryServerInterceptor(ctx)...),
 			)
-			if serverErr != nil {
-				l.Error("failed to create session store server", zap.Error(serverErr))
+			batonV1.RegisterBatonSessionServiceServer(grpcServer, sessionServer)
+			batonV1.RegisterBatonSourceCacheServiceServer(grpcServer, sourceCacheServer)
+			go func() {
+				<-ctx.Done()
+				grpcServer.GracefulStop()
+			}()
+			if serveErr := grpcServer.Serve(sessionListener); serveErr != nil {
+				l.Error("session/source-cache server stopped", zap.Error(serveErr))
 				return
 			}
 		}()
@@ -440,6 +483,7 @@ func (cw *wrapper) C(ctx context.Context) (types.ConnectorClient, error) {
 	cw.conn = conn
 	client := NewConnectorClient(ctx, cw.conn)
 	client.SetSessionStoreSetter(cw.SessionServer)
+	client.SetSourceCacheSetter(cw.SourceCacheServer)
 	cw.client = client
 
 	return client, nil