From b235e4a7dc7b07b2c63660665185ca6997d8e9d7 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Sat, 28 Feb 2026 17:51:34 -0600 Subject: [PATCH] update repomgr to support prevdata --- pkg/hold/pds/events.go | 126 ++++++++++++++++++++++++------------ pkg/hold/pds/events_test.go | 64 ++++++++++++++++++ pkg/hold/pds/import.go | 8 +++ pkg/hold/pds/repomgr.go | 101 ++++++++++++++++++++++++++--- 4 files changed, 246 insertions(+), 53 deletions(-) diff --git a/pkg/hold/pds/events.go b/pkg/hold/pds/events.go index 9d7ebbd..9fb9759 100644 --- a/pkg/hold/pds/events.go +++ b/pkg/hold/pds/events.go @@ -50,15 +50,16 @@ type HistoricalEvent struct { // RepoCommitEvent represents a #commit event in subscribeRepos type RepoCommitEvent struct { - Seq int64 `json:"seq" cborgen:"seq"` - Repo string `json:"repo" cborgen:"repo"` - Commit string `json:"commit" cborgen:"commit"` // CID string - Rev string `json:"rev" cborgen:"rev"` - Since *string `json:"since,omitempty" cborgen:"since,omitempty"` - Blocks []byte `json:"blocks" cborgen:"blocks"` // CAR slice bytes - Ops []*atproto.SyncSubscribeRepos_RepoOp `json:"ops" cborgen:"ops"` - Time string `json:"time" cborgen:"time"` - Type string `json:"$type" cborgen:"$type"` // Always "#commit" + Seq int64 `json:"seq" cborgen:"seq"` + Repo string `json:"repo" cborgen:"repo"` + Commit string `json:"commit" cborgen:"commit"` // CID string + Rev string `json:"rev" cborgen:"rev"` + Since *string `json:"since,omitempty" cborgen:"since,omitempty"` + PrevData string `json:"prevData,omitempty" cborgen:"prevData,omitempty"` // MST root CID string of previous commit + Blocks []byte `json:"blocks" cborgen:"blocks"` // CAR slice bytes + Ops []*atproto.SyncSubscribeRepos_RepoOp `json:"ops" cborgen:"ops"` + Time string `json:"time" cborgen:"time"` + Type string `json:"$type" cborgen:"$type"` // Always "#commit" } // NewEventBroadcaster creates a new event broadcaster with persistent storage @@ -155,6 +156,7 @@ func (b *EventBroadcaster) initSchema() error { commit_cid TEXT NOT NULL, rev TEXT NOT NULL, since_rev TEXT, + prev_data TEXT, repo_slice BLOB NOT NULL, ops_json TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP @@ -168,6 +170,9 @@ func (b *EventBroadcaster) initSchema() error { } } + // Migration: add prev_data column if missing (existing databases) + b.db.Exec("ALTER TABLE firehose_events ADD COLUMN prev_data TEXT") + // Load last sequence number from database var lastSeq sql.NullInt64 err := b.db.QueryRow("SELECT MAX(seq) FROM firehose_events").Scan(&lastSeq) @@ -522,12 +527,18 @@ func (b *EventBroadcaster) persistEvent(event *RepoCommitEvent) error { sinceRev = sql.NullString{String: *event.Since, Valid: true} } + // Get prev_data value (may be empty) + var prevData sql.NullString + if event.PrevData != "" { + prevData = sql.NullString{String: event.PrevData, Valid: true} + } + // Insert event query := ` - INSERT INTO firehose_events (seq, commit_cid, rev, since_rev, repo_slice, ops_json) - VALUES (?, ?, ?, ?, ?, ?) + INSERT INTO firehose_events (seq, commit_cid, rev, since_rev, prev_data, repo_slice, ops_json) + VALUES (?, ?, ?, ?, ?, ?, ?) ` - _, err = b.db.Exec(query, event.Seq, event.Commit, event.Rev, sinceRev, event.Blocks, opsJSON) + _, err = b.db.Exec(query, event.Seq, event.Commit, event.Rev, sinceRev, prevData, event.Blocks, opsJSON) return err } @@ -556,16 +567,23 @@ func (b *EventBroadcaster) convertToCommitEvent(event *RepoEvent, seq int64) *Re // Event.NewRoot is a cid.Cid, convert to string commitCID := event.NewRoot.String() + // Convert PrevData CID to string if present + var prevData string + if event.PrevData != nil { + prevData = event.PrevData.String() + } + return &RepoCommitEvent{ - Seq: seq, - Repo: b.holdDID, // Set to hold's DID - Commit: commitCID, - Rev: event.Rev, - Since: event.Since, - Blocks: event.RepoSlice, // CAR slice bytes - Ops: ops, - Time: time.Now().Format(time.RFC3339), - Type: "#commit", + Seq: seq, + Repo: b.holdDID, // Set to hold's DID + Commit: commitCID, + Rev: event.Rev, + Since: event.Since, + PrevData: prevData, + Blocks: event.RepoSlice, // CAR slice bytes + Ops: ops, + Time: time.Now().Format(time.RFC3339), + Type: "#commit", } } @@ -605,7 +623,7 @@ func (b *EventBroadcaster) backfillFromDatabase(sub *Subscriber, cursor int64) e // Query events where seq > cursor, ordered by seq // Include created_at to preserve original event timestamp query := ` - SELECT seq, commit_cid, rev, since_rev, repo_slice, ops_json, created_at + SELECT seq, commit_cid, rev, since_rev, prev_data, repo_slice, ops_json, created_at FROM firehose_events WHERE seq > ? ORDER BY seq ASC @@ -623,12 +641,13 @@ func (b *EventBroadcaster) backfillFromDatabase(sub *Subscriber, cursor int64) e commitCID string rev string sinceRev sql.NullString + prevData sql.NullString repoSlice []byte opsJSON []byte createdAt time.Time ) - if err := rows.Scan(&seq, &commitCID, &rev, &sinceRev, &repoSlice, &opsJSON, &createdAt); err != nil { + if err := rows.Scan(&seq, &commitCID, &rev, &sinceRev, &prevData, &repoSlice, &opsJSON, &createdAt); err != nil { slog.Error("Error scanning event row", "error", err) continue } @@ -646,16 +665,22 @@ func (b *EventBroadcaster) backfillFromDatabase(sub *Subscriber, cursor int64) e since = &sinceRev.String } + var prevDataStr string + if prevData.Valid { + prevDataStr = prevData.String + } + event := &RepoCommitEvent{ - Seq: seq, - Repo: b.holdDID, - Commit: commitCID, - Rev: rev, - Since: since, - Blocks: repoSlice, - Ops: ops, - Time: createdAt.Format(time.RFC3339), // Use original event time from database - Type: "#commit", + Seq: seq, + Repo: b.holdDID, + Commit: commitCID, + Rev: rev, + Since: since, + PrevData: prevDataStr, + Blocks: repoSlice, + Ops: ops, + Time: createdAt.Format(time.RFC3339), // Use original event time from database + Type: "#commit", } // Send to subscriber @@ -780,18 +805,33 @@ func convertToIndigoCommit(event *RepoCommitEvent) *atproto.SyncSubscribeRepos_C // Convert blocks to LexBytes blocks := lexutil.LexBytes(event.Blocks) + // Convert prevData CID string to LexLink if present + var prevDataLink *lexutil.LexLink + if event.PrevData != "" { + prevDataCID, err := cid.Decode(event.PrevData) + if err != nil { + slog.Warn("Failed to parse prevData CID", + "cid", event.PrevData, + "error", err) + } else { + link := lexutil.LexLink(prevDataCID) + prevDataLink = &link + } + } + return &atproto.SyncSubscribeRepos_Commit{ - Seq: event.Seq, - Repo: event.Repo, - Commit: commitLink, - Rev: event.Rev, - Since: event.Since, - Blocks: blocks, - Ops: event.Ops, - Time: event.Time, - Blobs: []lexutil.LexLink{}, // Empty for now, we don't track blob refs in our simplified model - Rebase: false, // DEPRECATED field - TooBig: false, // Not implementing tooBig for now + Seq: event.Seq, + Repo: event.Repo, + Commit: commitLink, + Rev: event.Rev, + Since: event.Since, + PrevData: prevDataLink, + Blocks: blocks, + Ops: event.Ops, + Time: event.Time, + Blobs: []lexutil.LexLink{}, // Empty for now, we don't track blob refs in our simplified model + Rebase: false, // DEPRECATED field + TooBig: false, // Not implementing tooBig for now } } diff --git a/pkg/hold/pds/events_test.go b/pkg/hold/pds/events_test.go index 9e37ddc..a1f8146 100644 --- a/pkg/hold/pds/events_test.go +++ b/pkg/hold/pds/events_test.go @@ -184,10 +184,12 @@ func TestConvertToCommitEvent(t *testing.T) { broadcaster := NewEventBroadcaster("did:web:hold.example.com", 10, "") testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke") + prevDataCID, _ := cid.Decode("bafyreie5737gdxlw5i64mnsc7x35mha5ee4w5vqnivxceag4pfd2mhqtsu") since := "prev-rev" event := &RepoEvent{ NewRoot: testCID, + PrevData: &prevDataCID, Rev: "test-rev-123", Since: &since, RepoSlice: []byte("test CAR data"), @@ -236,6 +238,11 @@ func TestConvertToCommitEvent(t *testing.T) { t.Errorf("Expected since=%s, got %v", since, commitEvent.Since) } + // Verify prevData is set + if commitEvent.PrevData != prevDataCID.String() { + t.Errorf("Expected prevData=%s, got %s", prevDataCID.String(), commitEvent.PrevData) + } + if string(commitEvent.Blocks) != "test CAR data" { t.Errorf("Expected blocks='test CAR data', got %s", string(commitEvent.Blocks)) } @@ -313,6 +320,63 @@ func TestConvertToCommitEvent_NoSince(t *testing.T) { if commitEvent.Since != nil { t.Errorf("Expected nil since, got %v", commitEvent.Since) } + + // PrevData should be empty when not set + if commitEvent.PrevData != "" { + t.Errorf("Expected empty prevData when not set, got %s", commitEvent.PrevData) + } +} + +// TestConvertToIndigoCommit_PrevData tests that prevData is correctly set on the indigo commit +func TestConvertToIndigoCommit_PrevData(t *testing.T) { + testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke") + prevDataCID, _ := cid.Decode("bafyreie5737gdxlw5i64mnsc7x35mha5ee4w5vqnivxceag4pfd2mhqtsu") + + event := &RepoCommitEvent{ + Seq: 1, + Repo: "did:web:hold.example.com", + Commit: testCID.String(), + Rev: "test-rev", + PrevData: prevDataCID.String(), + Blocks: []byte("test data"), + Ops: []*atproto.SyncSubscribeRepos_RepoOp{}, + Time: time.Now().Format(time.RFC3339), + Type: "#commit", + } + + indigoCommit := convertToIndigoCommit(event) + + // PrevData should be set + if indigoCommit.PrevData == nil { + t.Fatal("Expected non-nil PrevData on indigo commit") + } + + if cid.Cid(*indigoCommit.PrevData) != prevDataCID { + t.Errorf("Expected PrevData CID=%s, got %s", prevDataCID.String(), cid.Cid(*indigoCommit.PrevData).String()) + } +} + +// TestConvertToIndigoCommit_NoPrevData tests that empty prevData results in nil PrevData +func TestConvertToIndigoCommit_NoPrevData(t *testing.T) { + testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke") + + event := &RepoCommitEvent{ + Seq: 1, + Repo: "did:web:hold.example.com", + Commit: testCID.String(), + Rev: "test-rev", + Blocks: []byte("test data"), + Ops: []*atproto.SyncSubscribeRepos_RepoOp{}, + Time: time.Now().Format(time.RFC3339), + Type: "#commit", + } + + indigoCommit := convertToIndigoCommit(event) + + // PrevData should be nil when not set + if indigoCommit.PrevData != nil { + t.Errorf("Expected nil PrevData when not set, got %v", indigoCommit.PrevData) + } } // TestSetRepoEventHandler tests handler registration diff --git a/pkg/hold/pds/import.go b/pkg/hold/pds/import.go index 73e1a6c..74f43bf 100644 --- a/pkg/hold/pds/import.go +++ b/pkg/hold/pds/import.go @@ -120,6 +120,13 @@ func (p *HoldPDS) bulkImportRecords(ctx context.Context, records []bulkRecord) e return err } + // Capture previous MST root before commit overwrites it + var prevData *cid.Cid + if head.Defined() { + pd := r.DataCid() + prevData = &pd + } + ops := make([]RepoOp, 0, len(records)) for _, rec := range records { rpath := rec.Collection + "/" + rec.Rkey @@ -169,6 +176,7 @@ func (p *HoldPDS) bulkImportRecords(ctx context.Context, records []bulkRecord) e User: p.uid, OldRoot: oldroot, NewRoot: nroot, + PrevData: prevData, Rev: nrev, Since: &rev, Ops: ops, diff --git a/pkg/hold/pds/repomgr.go b/pkg/hold/pds/repomgr.go index 546c90b..a6e2b80 100644 --- a/pkg/hold/pds/repomgr.go +++ b/pkg/hold/pds/repomgr.go @@ -13,6 +13,7 @@ // - Added PutRecord method (lines 309-381) for creating records with explicit rkeys // (like CreateRecord but with specified rkey instead of auto-generated TID) // Based on streamplace/indigo implementation +// - Added prevData to support Sync 1.1 package pds import ( @@ -102,6 +103,7 @@ type RepoEvent struct { User models.Uid OldRoot *cid.Cid NewRoot cid.Cid + PrevData *cid.Cid // MST root CID of the previous commit (for firehose prevData field) Since *string Rev string RepoSlice []byte @@ -196,6 +198,13 @@ func (rm *RepoManager) CreateRecord(ctx context.Context, user models.Uid, collec return "", cid.Undef, err } + // Capture previous MST root before commit overwrites it + var prevData *cid.Cid + if head.Defined() { + pd := r.DataCid() + prevData = &pd + } + cc, tid, err := r.CreateRecord(ctx, collection, rec) if err != nil { return "", cid.Undef, err @@ -218,11 +227,12 @@ func (rm *RepoManager) CreateRecord(ctx context.Context, user models.Uid, collec if rm.events != nil { rm.events(ctx, &RepoEvent{ - User: user, - OldRoot: oldroot, - NewRoot: nroot, - Rev: nrev, - Since: &rev, + User: user, + OldRoot: oldroot, + NewRoot: nroot, + PrevData: prevData, + Rev: nrev, + Since: &rev, Ops: []RepoOp{{ Kind: EvtKindCreateRecord, Collection: collection, @@ -260,6 +270,13 @@ func (rm *RepoManager) UpdateRecord(ctx context.Context, user models.Uid, collec return cid.Undef, err } + // Capture previous MST root before commit overwrites it + var prevData *cid.Cid + if head.Defined() { + pd := r.DataCid() + prevData = &pd + } + rpath := collection + "/" + rkey cc, err := r.UpdateRecord(ctx, rpath, rec) if err != nil { @@ -297,6 +314,7 @@ func (rm *RepoManager) UpdateRecord(ctx context.Context, user models.Uid, collec User: user, OldRoot: oldroot, NewRoot: nroot, + PrevData: prevData, Rev: nrev, Since: &rev, Ops: []RepoOp{op}, @@ -334,6 +352,13 @@ func (rm *RepoManager) PutRecord(ctx context.Context, user models.Uid, collectio return "", cid.Undef, err } + // Capture previous MST root before commit overwrites it + var prevData *cid.Cid + if head.Defined() { + pd := r.DataCid() + prevData = &pd + } + rpath := collection + "/" + rkey cc, err := r.PutRecord(ctx, rpath, rec) if err != nil { @@ -371,6 +396,7 @@ func (rm *RepoManager) PutRecord(ctx context.Context, user models.Uid, collectio User: user, OldRoot: oldroot, NewRoot: nroot, + PrevData: prevData, Rev: nrev, Since: &rev, Ops: []RepoOp{op}, @@ -407,6 +433,13 @@ func (rm *RepoManager) UpsertRecord(ctx context.Context, user models.Uid, collec return "", cid.Undef, false, err } + // Capture previous MST root before commit overwrites it + var prevData *cid.Cid + if head.Defined() { + pd := r.DataCid() + prevData = &pd + } + rpath := collection + "/" + rkey // Check if record exists @@ -459,6 +492,7 @@ func (rm *RepoManager) UpsertRecord(ctx context.Context, user models.Uid, collec User: user, OldRoot: oldroot, NewRoot: nroot, + PrevData: prevData, Rev: nrev, Since: &rev, Ops: []RepoOp{op}, @@ -492,6 +526,13 @@ func (rm *RepoManager) DeleteRecord(ctx context.Context, user models.Uid, collec return err } + // Capture previous MST root before commit overwrites it + var prevData *cid.Cid + if head.Defined() { + pd := r.DataCid() + prevData = &pd + } + rpath := collection + "/" + rkey if err := r.DeleteRecord(ctx, rpath); err != nil { return err @@ -514,11 +555,12 @@ func (rm *RepoManager) DeleteRecord(ctx context.Context, user models.Uid, collec if rm.events != nil { rm.events(ctx, &RepoEvent{ - User: user, - OldRoot: oldroot, - NewRoot: nroot, - Rev: nrev, - Since: &rev, + User: user, + OldRoot: oldroot, + NewRoot: nroot, + PrevData: prevData, + Rev: nrev, + Since: &rev, Ops: []RepoOp{{ Kind: EvtKindDeleteRecord, Collection: collection, @@ -750,6 +792,16 @@ func (rm *RepoManager) handleExternalUserEventNoArchive(ctx context.Context, pds return fmt.Errorf("check repo sig: %w", err) } + // Capture previous MST root from old repo state if it exists + var prevData *cid.Cid + if ds.BaseCid().Defined() { + oldrepo, err := repo.OpenRepo(ctx, ds, ds.BaseCid()) + if err == nil { + pd := oldrepo.DataCid() + prevData = &pd + } + } + evtops := make([]RepoOp, 0, len(ops)) for _, op := range ops { parts := strings.SplitN(op.Path, "/", 2) @@ -809,6 +861,7 @@ func (rm *RepoManager) handleExternalUserEventNoArchive(ctx context.Context, pds User: uid, //OldRoot: prev, NewRoot: root, + PrevData: prevData, Rev: nrev, Since: since, Ops: evtops, @@ -846,12 +899,17 @@ func (rm *RepoManager) handleExternalUserEventArchive(ctx context.Context, pdsid } var skipcids map[cid.Cid]bool + var prevData *cid.Cid if ds.BaseCid().Defined() { oldrepo, err := repo.OpenRepo(ctx, ds, ds.BaseCid()) if err != nil { return fmt.Errorf("failed to check data root in old repo: %w", err) } + // Capture previous MST root for prevData + pd := oldrepo.DataCid() + prevData = &pd + // if the old commit has a 'prev', CalcDiff will error out while trying // to walk it. This is an old repo thing that is being deprecated. // This check is a temporary workaround until all repos get migrated @@ -933,6 +991,7 @@ func (rm *RepoManager) handleExternalUserEventArchive(ctx context.Context, pdsid User: uid, //OldRoot: prev, NewRoot: root, + PrevData: prevData, Rev: nrev, Since: since, Ops: evtops, @@ -967,6 +1026,13 @@ func (rm *RepoManager) BatchWrite(ctx context.Context, user models.Uid, writes [ return err } + // Capture previous MST root before commit overwrites it + var prevData *cid.Cid + if head.Defined() { + pd := r.DataCid() + prevData = &pd + } + ops := make([]RepoOp, 0, len(writes)) for _, w := range writes { switch { @@ -1054,6 +1120,7 @@ func (rm *RepoManager) BatchWrite(ctx context.Context, user models.Uid, writes [ User: user, OldRoot: oldroot, NewRoot: nroot, + PrevData: prevData, RepoSlice: rslice, Rev: nrev, Since: &rev, @@ -1095,6 +1162,19 @@ func (rm *RepoManager) ImportNewRepo(ctx context.Context, user models.Uid, repoD return fmt.Errorf("ImportNewRepo called with incorrect base") } + // Capture previous MST root before import overwrites it + var prevData *cid.Cid + if curhead.Defined() { + robs, err := rm.cs.ReadOnlySession(user) + if err == nil { + oldrepo, err := repo.OpenRepo(ctx, robs, curhead) + if err == nil { + pd := oldrepo.DataCid() + prevData = &pd + } + } + } + err = rm.processNewRepo(ctx, user, r, rev, func(ctx context.Context, root cid.Cid, finish func(context.Context, string) ([]byte, error), bs blockstore.Blockstore) error { r, err := repo.OpenRepo(ctx, bs, root) if err != nil { @@ -1139,6 +1219,7 @@ func (rm *RepoManager) ImportNewRepo(ctx context.Context, user models.Uid, repoD User: user, //OldRoot: oldroot, NewRoot: root, + PrevData: prevData, Rev: scom.Rev, Since: &currev, RepoSlice: slice,