feat(syncing): maintained RBSR fingerprint index for fast blob propagation

backend/api/documents/v3alpha/comments.go

@@ -510,9 +510,14 @@ var qGetCommentByCID = dqb.Str(`
 	WHERE (codec, multihash) = (:codec, :multihash)
 `)
 
+// qGetReplyCountByID counts distinct replying COMMENTS, not distinct reply
+// blobs: each edit of a comment is a separate blob carrying the same links, so
+// counting sources directly inflated the number by one per edit. Dedupe by the
+// source comment's identity (author + tsid).
 var qGetReplyCountByID = dqb.Str(`
-SELECT count(distinct source)
-FROM  blob_links bl
+SELECT count(DISTINCT src.author || ':' || (src.extra_attrs->>'tsid'))
+FROM blob_links bl
+JOIN structural_blobs src ON src.id = bl.source
 WHERE bl.target = (
    SELECT id
    FROM structural_blobs sb
@@ -522,7 +527,7 @@ WHERE bl.target = (
    AND sb.extra_attrs->>'tsid' = :tsid
 )
 AND bl.type IN ('comment/reply-parent', 'comment/thread-root')
-AND (SELECT sb.extra_attrs->>'deleted' FROM structural_blobs sb WHERE id = source) is not true
+AND src.extra_attrs->>'deleted' is not true
 `)
 
 var qListCommentVersions = dqb.Str(`

backend/api/documents/v3alpha/comments_test.go

@@ -95,6 +95,82 @@ func TestComments_Smoke(t *testing.T) {
 	testutil.StructsEqual(want, list).Compare(t, "comment list must match")
 }
 
+func TestGetCommentReplyCount_CountsCommentsNotVersions(t *testing.T) {
+	t.Parallel()
+
+	alice := newTestDocsAPI(t, "alice")
+	bob := coretest.NewTester("bob")
+	ctx := context.Background()
+	require.NoError(t, alice.keys.StoreKey(ctx, "bob", bob.Account))
+
+	homeDoc, err := alice.PublishDocumentChangeForTest(ctx, &apitest.DocumentChangeRequest{
+		SigningKeyName: "main",
+		Account:        alice.me.Account.PublicKey.String(),
+		Path:           "",
+		Changes: []*pb.DocumentChange{
+			{Op: &pb.DocumentChange_SetMetadata_{SetMetadata: &pb.DocumentChange_SetMetadata{Key: "title", Value: "Home"}}},
+		},
+	})
+	require.NoError(t, err)
+
+	root, err := alice.CreateComment(ctx, &pb.CreateCommentRequest{
+		SigningKeyName: "main",
+		TargetAccount:  alice.me.Account.PublicKey.String(),
+		TargetPath:     "",
+		TargetVersion:  homeDoc.Version,
+		Content:        []*pb.BlockNode{{Block: &pb.Block{Id: "b1", Type: "paragraph", Text: "Root"}}},
+	})
+	require.NoError(t, err)
+
+	reply, err := alice.CreateComment(ctx, &pb.CreateCommentRequest{
+		SigningKeyName: "bob",
+		TargetAccount:  alice.me.Account.PublicKey.String(),
+		TargetPath:     "",
+		TargetVersion:  homeDoc.Version,
+		ReplyParent:    root.Id,
+		Content:        []*pb.BlockNode{{Block: &pb.Block{Id: "b1", Type: "paragraph", Text: "Reply"}}},
+	})
+	require.NoError(t, err)
+
+	getCount := func() int64 {
+		res, err := alice.GetCommentReplyCount(ctx, &pb.GetCommentReplyCountRequest{Id: root.Id})
+		require.NoError(t, err)
+		return res.ReplyCount
+	}
+	require.EqualValues(t, 1, getCount())
+
+	// Editing the reply creates a new blob version carrying the same links; the
+	// count must stay per-comment, not per-version. ReplyParent must be passed
+	// on update, or the new version silently loses its thread linkage.
+	_, err = alice.UpdateComment(ctx, &pb.UpdateCommentRequest{
+		Comment: &pb.Comment{
+			Id:            reply.Id,
+			TargetAccount: alice.me.Account.PublicKey.String(),
+			TargetPath:    "",
+			TargetVersion: homeDoc.Version,
+			Author:        bob.Account.PublicKey.String(),
+			ReplyParent:   root.Id,
+			Content:       []*pb.BlockNode{{Block: &pb.Block{Id: "b1", Type: "paragraph", Text: "Reply, edited"}}},
+		},
+		SigningKeyName: "bob",
+	})
+	require.NoError(t, err)
+	require.EqualValues(t, 1, getCount(), "an edited reply must count once, not once per version")
+
+	// A deeper reply links to the thread root too: the root's count covers the
+	// whole thread, and every reply still counts exactly once.
+	_, err = alice.CreateComment(ctx, &pb.CreateCommentRequest{
+		SigningKeyName: "main",
+		TargetAccount:  alice.me.Account.PublicKey.String(),
+		TargetPath:     "",
+		TargetVersion:  homeDoc.Version,
+		ReplyParent:    reply.Id,
+		Content:        []*pb.BlockNode{{Block: &pb.Block{Id: "b1", Type: "paragraph", Text: "Deep reply"}}},
+	})
+	require.NoError(t, err)
+	require.EqualValues(t, 2, getCount(), "the thread root's count must include deep replies")
+}
+
 func TestListCommentsByAuthor(t *testing.T) {
 	t.Parallel()
 

backend/api/documents/v3alpha/contacts.go

@@ -118,7 +118,7 @@ func (srv *Server) ListContacts(ctx context.Context, in *documents.ListContactsR
 					COALESCE(res.owner, sb.author) as account_id,
 					ROW_NUMBER() OVER (
 						PARTITION BY sb.extra_attrs->>'tsid', COALESCE(res.owner, sb.author)
-						ORDER BY sb.ts DESC
+						ORDER BY sb.ts DESC, sb.id DESC
 					) as rn
 				FROM structural_blobs sb
 				LEFT JOIN resources res ON res.id = sb.resource
@@ -161,7 +161,7 @@ func (srv *Server) ListContacts(ctx context.Context, in *documents.ListContactsR
 					COALESCE(res.owner, sb.author) as account_id,
 					ROW_NUMBER() OVER (
 						PARTITION BY sb.extra_attrs->>'tsid', COALESCE(res.owner, sb.author)
-						ORDER BY sb.ts DESC
+						ORDER BY sb.ts DESC, sb.id DESC
 					) as rn
 				FROM structural_blobs sb
 				LEFT JOIN resources res ON res.id = sb.resource
@@ -298,7 +298,7 @@ func (srv *Server) GetContact(ctx context.Context, in *documents.GetContactReque
 			SELECT id FROM resources WHERE iri = 'hm://' || ?
 		)
 		AND sb.extra_attrs->>'tsid' = ?
-		ORDER BY sb.ts DESC
+		ORDER BY sb.ts DESC, sb.id DESC
 		LIMIT 1
 	`
 

backend/api/documents/v3alpha/contacts_test.go

@@ -725,3 +725,86 @@ func TestListContactsBySubjectWithTombstone(t *testing.T) {
 	require.NoError(t, err)
 	require.Len(t, resp.Contacts, 0, "Tombstoned contact should not appear when querying by account")
 }
+
+// TestListContactsTombstoneSameTimestamp is a regression guard for a flaky CI
+// failure: a deleted contact reappeared in ListContacts. Contact mutations are
+// stamped with millisecond wall-clock timestamps (structural_blobs.ts), so a
+// rapid create/update/delete sequence routinely lands the final write and its
+// tombstone in the same millisecond. When sb.ts ties, the "latest blob wins"
+// resolution (ROW_NUMBER OVER (... ORDER BY sb.ts DESC) for ListContacts, and
+// ORDER BY sb.ts DESC LIMIT 1 for GetContact) was non-deterministic and could
+// rank the non-deleted blob first, leaking the deleted contact back into the
+// list. The queries now break ties on sb.id (rowid / insert order), so the
+// tombstone — always Put last — wins. This test forces the tie explicitly by
+// giving the contact and its tombstone an identical timestamp, so it fails
+// deterministically against the un-fixed queries instead of relying on timing.
+func TestListContactsTombstoneSameTimestamp(t *testing.T) {
+	t.Parallel()
+
+	alice := newTestDocsAPI(t, "alice")
+	ctx := context.Background()
+
+	bob := coretest.NewTester("bob")
+
+	// Single timestamp shared by both the contact and its tombstone, rounded
+	// for blob encoding. This collapses sb.ts to a tie, leaving sb.id (insert
+	// order) as the only discriminator between the two blobs.
+	ts := time.Now().Round(blob.ClockPrecision)
+
+	contact, err := blob.NewContact(
+		alice.me.Account,
+		"", // empty TSID = generate new one
+		alice.me.Account.Principal(),
+		bob.Account.Principal(),
+		"Bob",
+		nil,
+		ts,
+	)
+	require.NoError(t, err)
+	require.NoError(t, alice.idx.Put(ctx, contact))
+
+	// Tombstone with the SAME timestamp as the contact, Put last.
+	tombstone, err := blob.NewContact(
+		alice.me.Account,
+		contact.TSID(),               // same TSID as the contact
+		alice.me.Account.Principal(), // same account
+		nil,                          // nil subject = tombstone
+		"",                           // empty name for tombstone
+		nil,
+		ts,
+	)
+	require.NoError(t, err)
+	require.NoError(t, alice.idx.Put(ctx, tombstone))
+
+	// ListContacts by account must not surface the tombstoned contact.
+	resp, err := alice.ListContacts(ctx, &documents.ListContactsRequest{
+		PageSize: 10,
+		Filter: &documents.ListContactsRequest_Account{
+			Account: alice.me.Account.PublicKey.String(),
+		},
+	})
+	require.NoError(t, err)
+	require.Len(t, resp.Contacts, 0, "Deleted contact must not appear when querying by account, even with a tied timestamp")
+
+	// ListContacts by subject must not surface it either.
+	resp, err = alice.ListContacts(ctx, &documents.ListContactsRequest{
+		PageSize: 10,
+		Filter: &documents.ListContactsRequest_Subject{
+			Subject: bob.Account.PublicKey.String(),
+		},
+	})
+	require.NoError(t, err)
+	require.Len(t, resp.Contacts, 0, "Deleted contact must not appear when querying by subject, even with a tied timestamp")
+
+	// GetContact must report NotFound.
+	contactID := blob.RecordID{
+		Authority: alice.me.Account.Principal(),
+		TSID:      contact.TSID(),
+	}.String()
+	_, err = alice.GetContact(ctx, &documents.GetContactRequest{
+		Id: contactID,
+	})
+	require.Error(t, err)
+	st, _ := status.FromError(err)
+	require.Equal(t, codes.NotFound, st.Code(), "GetContact must return NotFound for a tombstoned contact with a tied timestamp")
+}

backend/blob/blob_capability.go

@@ -166,5 +166,5 @@ func indexCapability(ictx *indexingCtx, _ int64, eb Encoded[*Capability]) error
 	// capability change (e.g. revocation) without further code changes.
 	ictx.writerCache.clear()
 
-	return reindexStashedBlobs(ictx.mustTrackUnreads, ictx.conn, stashReasonPermissionDenied, v.Delegate.String(), ictx.blockStore, ictx.log, ictx.writerCache)
+	return reindexStashedBlobs(ictx.mustTrackUnreads, ictx.conn, stashReasonPermissionDenied, v.Delegate.String(), ictx.blockStore, ictx.log, ictx.writerCache, ictx.hookIDs)
 }

backend/blob/blob_capability_test.go

@@ -5,6 +5,7 @@ import (
 	"seed/backend/storage"
 	"seed/backend/util/cclock"
 	"seed/backend/util/must"
+	"seed/backend/util/sqlite"
 	"seed/backend/util/sqlite/sqlitex"
 	"testing"
 
@@ -41,8 +42,58 @@ func TestOutOfOrderCapability(t *testing.T) {
 	require.Equal(t, 0, countStashedBlobs(t, db), "must have no stashed blobs when ref is indexed before the needed capability")
 }
 
+// TestOutOfOrderCapability_FiresIndexedHookForUnstashed guards that the indexed
+// hook (which maintains the RBSR index) sees blobs re-indexed by the unstash
+// cascade, not just the top-level blob of each Put. A late capability unstashes
+// the previously permission-denied ref; without threading the cascade ids to the
+// hook, the maintained RBSR index would go stale-short for that ref.
+func TestOutOfOrderCapability_FiresIndexedHookForUnstashed(t *testing.T) {
+	alice := coretest.NewTester("alice").Account
+	bob := coretest.NewTester("bob").Account
+	db := storage.MakeTestDB(t)
+	idx, err := OpenIndex(t.Context(), db, zap.NewNop())
+	require.NoError(t, err)
+
+	var hookIDs []int64
+	idx.SetIndexedHook(func(_ *sqlite.Conn, ids []int64) error {
+		hookIDs = append(hookIDs, ids...)
+		return nil
+	})
+
+	clock := cclock.New()
+	change, err := NewChange(bob, cid.Undef, nil, 0, ChangeBody{
+		Ops: []OpMap{
+			must.Do2(NewOpSetKey("name", "Hello")),
+		},
+	}, clock.MustNow())
+	require.NoError(t, err)
+
+	ref, err := NewRef(bob, 0, change.CID, alice.Principal(), "", []cid.Cid{change.CID}, clock.MustNow(), VisibilityPublic)
+	require.NoError(t, err)
+
+	cpb, err := NewCapability(alice, bob.Principal(), alice.Principal(), "", "WRITER", "", clock.MustNow())
+	require.NoError(t, err)
+
+	// ref arrives before the capability that authorizes bob, so it is stashed.
+	require.NoError(t, idx.Put(t.Context(), ref))
+	require.NoError(t, idx.Put(t.Context(), change))
+	// The capability unstashes ref via the reindex cascade.
+	require.NoError(t, idx.Put(t.Context(), cpb))
+
+	require.Equal(t, 0, countStashedBlobs(t, db), "capability must unstash the ref")
+
+	refID := blobIDForCID(t, db, ref.CID)
+	require.Contains(t, hookIDs, refID, "indexed hook must fire for the blob unstashed by the capability cascade")
+}
+
 func countStashedBlobs(t *testing.T, db *sqlitex.Pool) int {
 	count, err := sqlitex.QueryOnePool[int](t.Context(), db, "SELECT count() FROM stashed_blobs")
 	require.NoError(t, err)
 	return count
 }
+
+func blobIDForCID(t *testing.T, db *sqlitex.Pool, c cid.Cid) int64 {
+	id, err := sqlitex.QueryOnePool[int64](t.Context(), db, "SELECT id FROM blobs WHERE multihash = ?", []byte(c.Hash()))
+	require.NoError(t, err)
+	return id
+}

backend/blob/blob_change.go

@@ -534,7 +534,7 @@ func indexChange(ictx *indexingCtx, id int64, eb Encoded[*Change]) error {
 		return err
 	}
 
-	if err := reindexStashedBlobs(ictx.mustTrackUnreads, ictx.conn, stashReasonFailedPrecondition, c.String(), ictx.blockStore, ictx.log, ictx.writerCache); err != nil {
+	if err := reindexStashedBlobs(ictx.mustTrackUnreads, ictx.conn, stashReasonFailedPrecondition, c.String(), ictx.blockStore, ictx.log, ictx.writerCache, ictx.hookIDs); err != nil {
 		return err
 	}
 

backend/blob/blob_comment.go

@@ -326,7 +326,7 @@ func indexComment(ictx *indexingCtx, id int64, eb Encoded[*Comment]) error {
 
 	// If the comment we've just indexed was a reply parent of another comment we've seen before,
 	// we need to reindex those comments.
-	if err := reindexStashedBlobs(ictx.mustTrackUnreads, ictx.conn, stashReasonFailedPrecondition, c.String(), ictx.blockStore, ictx.log, ictx.writerCache); err != nil {
+	if err := reindexStashedBlobs(ictx.mustTrackUnreads, ictx.conn, stashReasonFailedPrecondition, c.String(), ictx.blockStore, ictx.log, ictx.writerCache, ictx.hookIDs); err != nil {
 		return err
 	}