[experimental] Sync Replay (etaging)

cmd/baton-github/main.go

@@ -14,5 +14,8 @@ var version = "dev"
 
 func main() {
 	ctx := context.Background()
-	config.RunConnector(ctx, "baton-github", version, cfg.Config, connector.NewLambdaConnector, connectorrunner.WithSessionStoreEnabled())
+	config.RunConnector(ctx, "baton-github", version, cfg.Config, connector.NewLambdaConnector,
+		connectorrunner.WithSessionStoreEnabled(),
+		connectorrunner.WithKeepPreviousSyncC1Z(),
+	)
 }

go.mod

@@ -2,6 +2,8 @@ module github.com/conductorone/baton-github
 
 go 1.25.2
 
+replace github.com/conductorone/baton-sdk => /Users/matt.kaniaris/work/baton-sdk-2
+
 require (
 	github.com/conductorone/baton-sdk v0.17.0
 	github.com/deckarep/golang-set/v2 v2.9.0

pkg/config/config.go

@@ -94,6 +94,13 @@ var Config = field.NewConfiguration(
 		syncSecrets,
 		omitArchivedRepositories,
 		directCollaboratorsOnly,
+		// Default to a small worker pool: the source-cache warm path fans
+		// member pages out via SpawnCursors, and cold syncs parallelize
+		// per-resource work. 4 keeps sustained request rate well under
+		// GitHub's secondary (points-per-minute) limits; the SDK limiter
+		// backs off on 429/secondary responses regardless. --workers/
+		// BATON_WORKERS still override.
+		field.WithConnectorDefault(field.WorkerCountField, 4),
 	},
 	field.WithConnectorDisplayName("GitHub v2"),
 	field.WithHelpUrl("/docs/baton/github-v2"),

pkg/connector/connector.go

@@ -3,7 +3,9 @@ package connector
 import (
 	"context"
 	"crypto/rsa"
+	"crypto/sha256"
 	"crypto/x509"
+	"encoding/hex"
 	"encoding/pem"
 	"errors"
 	"fmt"
@@ -127,11 +129,18 @@ type GitHub struct {
 	omitArchivedRepositories bool
 	directCollaboratorsOnly  bool
 	enterprises              []string
+	// authScope is a stable, non-secret identity for the configured
+	// credential, mixed into source-cache scope signatures so two
+	// different credentials (with different member visibility) never
+	// share cached scopes. PATs use a token digest; GitHub Apps use the
+	// stable installation identity, which survives hourly installation
+	// token rotation.
+	authScope string
 }
 
 func (gh *GitHub) ResourceSyncers(ctx context.Context) []connectorbuilder.ResourceSyncerV2 {
 	resourceSyncers := []connectorbuilder.ResourceSyncerV2{
-		OrgBuilder(gh.client, gh.appClient, gh.orgCache, gh.orgs, gh.syncSecrets),
+		OrgBuilder(gh.client, gh.appClient, gh.orgCache, gh.orgs, gh.syncSecrets, gh.authScope),
 		TeamBuilder(gh.client, gh.orgCache, gh.directCollaboratorsOnly),
 		UserBuilder(gh.client, gh.graphqlClient, gh.orgCache, gh.orgs, gh.customClient, gh.enterprises),
 		RepositoryBuilder(gh.client, gh.orgCache, gh.omitArchivedRepositories, gh.directCollaboratorsOnly),
@@ -254,7 +263,7 @@ func (gh *GitHub) Validate(ctx context.Context) (annotations.Annotations, error)
 				zap.Error(err))
 		}
 	}
-	return nil, nil
+	return sourceCacheCapabilityAnnotations(), nil
 }
 
 func (gh *GitHub) validateAppCredentials(ctx context.Context) (annotations.Annotations, error) {
@@ -279,7 +288,13 @@ func (gh *GitHub) validateAppCredentials(ctx context.Context) (annotations.Annot
 		}
 	}
 
-	return nil, nil
+	return sourceCacheCapabilityAnnotations(), nil
+}
+
+func sourceCacheCapabilityAnnotations() annotations.Annotations {
+	return annotations.New(v2.SourceCacheCapability_builder{
+		Mode: v2.SourceCacheCapability_MODE_READ_WRITE,
+	}.Build())
 }
 
 // newGitHubClient returns a new GitHub API client authenticated with an access token via oauth2.
@@ -346,9 +361,20 @@ func newWithGithubPAT(ctx context.Context, ghc *cfg.Github) (*GitHub, error) {
 		syncSecrets:              ghc.SyncSecrets,
 		omitArchivedRepositories: ghc.OmitArchivedRepositories,
 		directCollaboratorsOnly:  ghc.DirectCollaboratorsOnly,
+		authScope:                patAuthScope(ghc.Token),
 	}, nil
 }
 
+// patAuthScope derives a stable, non-secret identity from a personal
+// access token for source-cache scope isolation. The digest is truncated
+// (it only needs to distinguish credentials, not resist collision
+// attacks) and never logged alongside anything that would identify it as
+// token-derived material.
+func patAuthScope(token string) string {
+	sum := sha256.Sum256([]byte("baton-github-auth-scope|" + token))
+	return "pat:" + hex.EncodeToString(sum[:8])
+}
+
 func newWithGithubApp(ctx context.Context, ghc *cfg.Github) (*GitHub, error) {
 	jwttoken, err := getJWTToken(ghc.AppId, string(ghc.AppPrivatekeyPath))
 	if err != nil {
@@ -433,6 +459,10 @@ func newWithGithubApp(ctx context.Context, ghc *cfg.Github) (*GitHub, error) {
 		syncSecrets:              ghc.SyncSecrets,
 		omitArchivedRepositories: ghc.OmitArchivedRepositories,
 		directCollaboratorsOnly:  ghc.DirectCollaboratorsOnly,
+		// The installation identity is stable across the hourly
+		// installation-token rotation, so cached scopes survive rotation
+		// (a per-token digest would go cold every hour).
+		authScope: fmt.Sprintf("app:%s:install:%d", ghc.AppId, installation.GetID()),
 	}
 	return gh, nil
 }

pkg/connector/org.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"net/url"
 	"strconv"
 	"strings"
 
@@ -39,6 +40,9 @@ type orgResourceType struct {
 	orgs         map[string]struct{}
 	orgCache     *orgNameCache
 	syncSecrets  bool
+	// authScope isolates source-cache scopes per credential; GitHub ETag
+	// visibility is credential-specific (see source_cache.go).
+	authScope string
 }
 
 func organizationResource(
@@ -183,15 +187,120 @@ func (o *orgResourceType) orgRoleGrant(roleName string, org *v2.Resource, princi
 	}))
 }
 
+// orgMembersPageURL builds one page's URL with a fixed query-param order
+// so the same logical page hashes to the same scope every sync.
+func orgMembersPageURL(orgName, role string, page int) string {
+	q := url.Values{}
+	q.Set("page", strconv.Itoa(page))
+	q.Set("per_page", strconv.Itoa(maxPageSize))
+	q.Set("role", role)
+	return fmt.Sprintf("orgs/%v/members?%s", orgName, q.Encode())
+}
+
+// listMembersConditional fetches one page of the org member listing for
+// one role, with source-cache revalidation (see source_cache.go for the
+// scope model, the horizon clamp, and token-carried validators).
+func (o *orgResourceType) listMembersConditional(
+	ctx context.Context,
+	orgName string,
+	role string,
+	cursor pagedListCursor,
+	opts resourceSdk.SyncOpAttrs,
+) (*conditionalPage, error) {
+	pageURL := orgMembersPageURL(orgName, role, cursor.Page)
+	return listUsersPageConditional(ctx, o.client, opts.SourceCache, o.authScope, "org-members",
+		pageURL, cursor, maxPageSize)
+}
+
+// cursorForResolvedPage builds a page cursor carrying the page's
+// validator resolution from the origin's batched walk, so the page does
+// no lookup of its own (zero ask/answer bounces on the Lambda
+// continuation). Unqueried pages get an unresolved cursor (single-lookup
+// fallback).
+func cursorForResolvedPage(page int, horizon int, resolved map[int]resolvedPage) pagedListCursor {
+	c := pagedListCursor{Page: page, Horizon: horizon}
+	r, ok := resolved[page]
+	if !ok {
+		return c
+	}
+	if r.Found {
+		c.Resolution = cursorValidatorHit
+		c.Etag = r.Validator
+	} else {
+		c.Resolution = cursorValidatorMiss
+	}
+	return c
+}
+
+// spawnMemberPageCursors resolves one role collection's stored page chain
+// (a single batched lookup — one ask bounce on the Lambda continuation)
+// and computes the spawn set: on a warm sync the connector emits one
+// sibling cursor per stored page 2..horizon and the SDK worker pool
+// revalidates them concurrently instead of one round trip per page
+// serially. Each token is a self-contained single-state pagination bag
+// (the resource identity rides the spawned action) carrying the page's
+// validator, so siblings do no lookups of their own.
+//
+// Returns the page-1 cursor (with its own resolution embedded), the
+// marshaled sibling tokens (nil when there is no stored chain — cold
+// syncs paginate serially exactly as before), or an error, which may
+// wrap sourcecache.ErrLookupDeferred and must propagate.
+func (o *orgResourceType) spawnMemberPageCursors(
+	ctx context.Context,
+	orgName string,
+	role string,
+	opts resourceSdk.SyncOpAttrs,
+) (pagedListCursor, []string, error) {
+	resolved, horizon, err := resolveStoredPageChain(ctx, o.client, opts.SourceCache, o.authScope, "org-members",
+		func(page int) string { return orgMembersPageURL(orgName, role, page) })
+	if err != nil {
+		return pagedListCursor{}, nil, err
+	}
+	if horizon < 2 {
+		// One stored page or none: nothing to fan out. Page 1 still
+		// carries its own resolution (and the real horizon, for the
+		// past-horizon miss propagation on chains) so no further lookups
+		// happen.
+		return cursorForResolvedPage(1, horizon, resolved), nil, nil
+	}
+	tokens := make([]string, 0, horizon-1)
+	for p := 2; p <= horizon; p++ {
+		cursorTok, err := cursorForResolvedPage(p, horizon, resolved).marshal()
+		if err != nil {
+			return pagedListCursor{}, nil, err
+		}
+		sibling := &pagination.Bag{}
+		sibling.Push(pagination.PageState{
+			ResourceTypeID: role,
+			Token:          cursorTok,
+		})
+		tok, err := sibling.Marshal()
+		if err != nil {
+			return pagedListCursor{}, nil, err
+		}
+		tokens = append(tokens, tok)
+	}
+	return cursorForResolvedPage(1, horizon, resolved), tokens, nil
+}
+
 func (o *orgResourceType) Grants(
 	ctx context.Context,
 	resource *v2.Resource,
 	opts resourceSdk.SyncOpAttrs,
 ) ([]*v2.Grant, *resourceSdk.SyncOpResults, error) {
-	bag, page, err := parsePageToken(opts.PageToken.Token, resource.Id)
-	if err != nil {
+	// The role states carry a JSON pagedListCursor token, so the bag is
+	// unmarshaled directly instead of through parsePageToken (which
+	// assumes integer tokens).
+	bag := &pagination.Bag{}
+	if err := bag.Unmarshal(opts.PageToken.Token); err != nil {
 		return nil, nil, err
 	}
+	if bag.Current() == nil {
+		bag.Push(pagination.PageState{
+			ResourceTypeID: resource.Id.ResourceType,
+			ResourceID:     resource.Id.Resource,
+		})
+	}
 
 	var (
 		reqAnnos  annotations.Annotations
@@ -214,33 +323,66 @@ func (o *orgResourceType) Grants(
 		if err != nil {
 			return nil, nil, err
 		}
-		listOpts := github.ListMembersOptions{
-			Role: rId,
-			ListOptions: github.ListOptions{
-				Page:    page,
-				PerPage: maxPageSize,
-			},
+		cursor, err := parsePagedListCursor(bag.PageToken())
+		if err != nil {
+			return nil, nil, err
+		}
+
+		// The collection's first call resolves the stored chain in one
+		// batched lookup and fans out: spawn one sibling cursor per stored
+		// page (each carrying its validator, so siblings do no lookups),
+		// and adopt the horizon so this page's own chain never continues
+		// into sibling-owned pages. Spawned and chained cursors carry a
+		// resolution/horizon already and never re-spawn.
+		var spawnTokens []string
+		if cursor.Page == 1 && cursor.Horizon == 0 && cursor.Resolution == "" {
+			cursor, spawnTokens, err = o.spawnMemberPageCursors(ctx, orgName, rId, opts)
+			if err != nil {
+				return nil, nil, err
+			}
 		}
-		users, resp, err := o.client.Organizations.ListMembers(ctx, orgName, &listOpts)
+
+		pageRes, err := o.listMembersConditional(ctx, orgName, rId, cursor, opts)
 		if err != nil {
+			var resp *github.Response
+			if pageRes != nil {
+				resp = pageRes.Resp
+			}
 			if isNotFoundError(resp) {
 				return nil, nil, uhttp.WrapErrors(codes.NotFound, fmt.Sprintf("org: %s not found", orgName))
 			}
-			return nil, nil, wrapGitHubError(err, resp, "github-connector: failed to list org members")
+			return nil, nil, err
 		}
-
-		var nextPage string
-		nextPage, reqAnnos, err = parseResp(resp)
-		if err != nil {
-			return nil, nil, fmt.Errorf("github-connectorv2: failed to parse response: %w", err)
+		reqAnnos = pageRes.Annos
+		if len(spawnTokens) > 0 {
+			reqAnnos.Append(v2.SpawnCursors_builder{
+				PageTokens: spawnTokens,
+			}.Build())
 		}
 
-		err = bag.Next(nextPage)
+		nextToken := ""
+		if pageRes.NextPage != 0 {
+			nextCursor := pagedListCursor{Page: pageRes.NextPage, Horizon: cursor.Horizon}
+			// Chained pages beyond the stored chain are known misses (the
+			// chain is contiguous): a cold page's continuation, and the
+			// probe past the horizon. Carrying the miss skips their
+			// lookups; if ever wrong it merely forgoes a conditional
+			// request and fetches cold — never stale.
+			if cursor.Resolution == cursorValidatorMiss ||
+				(cursor.Horizon > 0 && pageRes.NextPage > cursor.Horizon) {
+				nextCursor.Resolution = cursorValidatorMiss
+			}
+			nextToken, err = nextCursor.marshal()
+			if err != nil {
+				return nil, nil, err
+			}
+		}
+		err = bag.Next(nextToken)
 		if err != nil {
 			return nil, nil, err
 		}
 
-		for _, user := range users {
+		for _, user := range pageRes.Users {
 			ur, err := userResource(ctx, user, user.GetEmail(), nil)
 			if err != nil {
 				return nil, nil, err
@@ -257,7 +399,7 @@ func (o *orgResourceType) Grants(
 		)
 	}
 
-	pageToken, err = bag.Marshal()
+	pageToken, err := bag.Marshal()
 	if err != nil {
 		return nil, nil, err
 	}
@@ -414,7 +556,7 @@ func (o *orgResourceType) Revoke(ctx context.Context, grant *v2.Grant) (annotati
 	return nil, nil
 }
 
-func OrgBuilder(client, appClient *github.Client, orgCache *orgNameCache, orgs []string, syncSecrets bool) *orgResourceType {
+func OrgBuilder(client, appClient *github.Client, orgCache *orgNameCache, orgs []string, syncSecrets bool, authScope string) *orgResourceType {
 	orgMap := make(map[string]struct{})
 
 	for _, o := range orgs {
@@ -428,6 +570,7 @@ func OrgBuilder(client, appClient *github.Client, orgCache *orgNameCache, orgs [
 		appClient:    appClient,
 		orgCache:     orgCache,
 		syncSecrets:  syncSecrets,
+		authScope:    authScope,
 	}
 }
 

pkg/connector/org_test.go

@@ -24,7 +24,7 @@ func TestOrganization(t *testing.T) {
 
 		githubClient := github.NewClient(mgh.Server())
 		cache := newOrgNameCache(githubClient)
-		client := OrgBuilder(githubClient, nil, cache, nil, false)
+		client := OrgBuilder(githubClient, nil, cache, nil, false, "test-auth")
 
 		organization, _ := organizationResource(ctx, githubOrganization, nil, false)
 		user, _ := userResource(ctx, githubUser, *githubUser.Email, nil)