diff --git a/pkg/atproto/atproto.go b/pkg/atproto/atproto.go index f5ac04e1..e5799aa1 100644 --- a/pkg/atproto/atproto.go +++ b/pkg/atproto/atproto.go @@ -1,19 +1,17 @@ package atproto import ( - "bytes" "context" "fmt" "net" + "sync" "time" "github.com/patrickmn/go-cache" "github.com/bluesky-social/indigo/atproto/identity" "github.com/bluesky-social/indigo/atproto/syntax" - "github.com/bluesky-social/indigo/repo" "github.com/bluesky-social/indigo/xrpc" - "github.com/ipfs/go-cid" "go.opentelemetry.io/otel" "stream.place/streamplace/pkg/aqhttp" "stream.place/streamplace/pkg/comatproto" @@ -32,18 +30,18 @@ func (atsync *ATProtoSynchronizer) SyncBlueskyRepoCached(ctx context.Context, ha if err != nil { return nil, fmt.Errorf("failed to get repo for %s: %w", handle, err) } - if repo != nil { + // An empty Version means the row is a placeholder written at the start of a + // backfill that never finished, so the repo is only partially indexed. Fall + // through and sync it again -- unless the backfill that wrote it is still + // running in this process, in which case the placeholder is exactly what + // the caller should see (records being indexed right now call back in here). + if repo != nil && (repo.Version != "" || syncInFlight(repo.DID)) { return repo, nil } return atsync.SyncBlueskyRepo(ctx, handle, atsync.Model) } -type mstNode struct { - rkey syntax.RecordKey - collection syntax.NSID -} - func (atsync *ATProtoSynchronizer) SyncBlueskyRepo(ctx context.Context, handle string, mod model.Model) (*model.Repo, error) { ident, err := atsync.resolveIdent(ctx, handle, true) if err != nil { @@ -56,14 +54,25 @@ func (atsync *ATProtoSynchronizer) SyncBlueskyRepo(ctx context.Context, handle s handleLock.Lock() defer handleLock.Unlock() - rev := "" + // Tell re-entrant callers (handleCreateUpdate syncs the repos it sees + // records from) that this DID's placeholder row is being filled in right + // now, so they take the placeholder instead of recursing into the per-DID + // lock we are holding. + defer markSyncInFlight(ident.DID.String())() + oldRepo, err := mod.GetRepo(ident.DID.String()) if err != nil { return nil, fmt.Errorf("failed to get DID record for %s: %w", ident.DID.String(), err) } - if oldRepo != nil { + if oldRepo != nil && oldRepo.Version != "" { log.Log(ctx, "found existing DID record", "did", oldRepo.DID, "version", oldRepo.Version) return oldRepo, nil + } + if oldRepo != nil { + // A placeholder from a backfill that never finished: the repo is + // half-indexed, so sync it again rather than leaving it that way + // forever. The placeholder row is already there, don't rewrite it. + log.Log(ctx, "found incomplete DID record, re-syncing", "did", oldRepo.DID) } else { // create an empty repo while we sync. this is useful because we'll start monitoring the firehose for // any new follows and such from this user while we're syncing, which can take a long time @@ -84,7 +93,6 @@ func (atsync *ATProtoSynchronizer) SyncBlueskyRepo(ctx context.Context, handle s } log.Log(ctx, "resolved bluesky identity", "did", ident.DID, "handle", ident.Handle, "pds", ident.PDSEndpoint()) - pdsLock := pdsLocks.GetLock(ident.PDSEndpoint()) xrpcc := xrpc.Client{ Host: ident.PDSEndpoint(), Client: &aqhttp.Client, @@ -92,73 +100,22 @@ func (atsync *ATProtoSynchronizer) SyncBlueskyRepo(ctx context.Context, handle s if xrpcc.Host == "" { return nil, fmt.Errorf("no PDS endpoint found for Bluesky identity %s", handle) } - pdsLock.Lock() - repoBytes, err := SyncGetRepo(ctx, &xrpcc, ident.DID.String(), rev) - pdsLock.Unlock() - if err != nil { - return nil, fmt.Errorf("failed to fetch repo for %s from PDS %s: %w", ident.DID.String(), xrpcc.Host, err) - } - - // uncomment for saving new test cases: - - // timestamp := time.Now().Unix() - // filename := fmt.Sprintf("%d.base64", timestamp) - // encodedBytes := base64.URLEncoding.EncodeToString(repoBytes) - // err = os.WriteFile(filename, []byte(encodedBytes), 0644) - // if err != nil { - // return nil, fmt.Errorf("failed to write encoded repo bytes to file: %w", err) - // } - - log.Debug(ctx, "got diff", "bytes", len(repoBytes)) - - r, err := repo.ReadRepoFromCar(ctx, bytes.NewReader(repoBytes)) - if err != nil { - return nil, fmt.Errorf("failed to parse repo CAR data for %s: %w", ident.DID.String(), err) - } - // extract DID from repo commit - sc := r.SignedCommit() - signerDID, err := syntax.ParseDID(sc.Did) - if err != nil { - return nil, fmt.Errorf("invalid DID in repo commit for %s: %w", ident.DID.String(), err) - } - if signerDID != ident.DID { - return nil, fmt.Errorf("signer DID %s does not match identity %s", signerDID, ident.DID.String()) - } - - err = r.ForEach(ctx, "", func(k string, v cid.Cid) error { - nsid, rkey, err := syntax.ParseRepoPath(k) - if err != nil { - log.Warn(ctx, "failed to parse repo path", "k", k, "err", err) - return fmt.Errorf("could not parse repo path %s: %w", k, err) - } - _, bs, err := r.GetRecordBytes(ctx, k) - if err != nil { - log.Warn(ctx, "failed to get record bytes", "k", k, "rkey", rkey, "err", err) - return fmt.Errorf("could not retrieve record bytes for %s (rkey: %s): %w", k, rkey, err) - } - log.Debug(ctx, "record type", "key", k, "type", nsid.String()) - err = atsync.handleCreateUpdate(ctx, signerDID.String(), rkey, bs, v.String(), nsid, false, true) - if err != nil { - log.Warn(ctx, "failed to handle create update", "err", err) - // invalid CBOR and stuff should get ignored, so - // return fmt.Errorf("failed to process record update for %s (type: %s): %w", k, nsid.String(), err) - } - return nil - }) + rev, rootCID, err := atsync.backfillRepo(ctx, ident, &xrpcc) if err != nil { - return nil, fmt.Errorf("failed to iterate over repo: %w", err) + return nil, err } newRepo := model.Repo{ DID: ident.DID.String(), PDS: ident.PDSEndpoint(), - Version: sc.Rev, + Version: rev, + RootCID: rootCID, Handle: ident.Handle.String(), } err = mod.UpdateRepo(&newRepo) if err != nil { - return nil, fmt.Errorf("failed to update DID record for %s: %w", sc.Did, err) + return nil, fmt.Errorf("failed to update DID record for %s: %w", ident.DID.String(), err) } err = atsync.StatefulDB.AddRepo(ident.DID.String()) if err != nil { @@ -168,6 +125,37 @@ func (atsync *ATProtoSynchronizer) SyncBlueskyRepo(ctx context.Context, handle s return &newRepo, nil } +// syncsInFlight holds the DIDs whose backfill is running in this process right +// now. A placeholder repo row (empty Version) otherwise means "incomplete, +// re-sync me", which would be wrong -- and, since indexing a record can call +// back into SyncBlueskyRepoCached for the same DID, deadlock on the per-DID +// lock -- while the backfill that wrote it is still going. +var syncsInFlight = struct { + sync.Mutex + dids map[string]int +}{dids: map[string]int{}} + +func markSyncInFlight(did string) func() { + syncsInFlight.Lock() + syncsInFlight.dids[did]++ + syncsInFlight.Unlock() + return func() { + syncsInFlight.Lock() + defer syncsInFlight.Unlock() + if syncsInFlight.dids[did] <= 1 { + delete(syncsInFlight.dids, did) + return + } + syncsInFlight.dids[did]-- + } +} + +func syncInFlight(did string) bool { + syncsInFlight.Lock() + defer syncsInFlight.Unlock() + return syncsInFlight.dids[did] > 0 +} + func (atsync *ATProtoSynchronizer) RefreshIdentity(ctx context.Context, did string) (*identity.Identity, error) { id, err := atsync.resolveIdent(ctx, did, false) if err != nil { @@ -178,6 +166,18 @@ func (atsync *ATProtoSynchronizer) RefreshIdentity(ctx context.Context, did stri PDS: id.PDSEndpoint(), Handle: id.Handle.String(), } + // UpdateRepo writes every column, so carry the sync state over: blanking + // Version here would mark the repo as never-backfilled and (now that an + // empty Version means "re-sync me") make every identity event trigger a + // pointless full re-index. + oldRepo, err := atsync.Model.GetRepo(id.DID.String()) + if err != nil { + return nil, fmt.Errorf("failed to get repo: %w", err) + } + if oldRepo != nil { + newRepo.Version = oldRepo.Version + newRepo.RootCID = oldRepo.RootCID + } err = atsync.Model.UpdateRepo(&newRepo) if err != nil { return nil, fmt.Errorf("failed to update repo: %w", err) diff --git a/pkg/atproto/backfill_walk.go b/pkg/atproto/backfill_walk.go new file mode 100644 index 00000000..c2e6721e --- /dev/null +++ b/pkg/atproto/backfill_walk.go @@ -0,0 +1,240 @@ +package atproto + +import ( + "bytes" + "context" + "errors" + "fmt" + "net/http" + "strings" + "sync" + + "github.com/bluesky-social/indigo/atproto/identity" + "github.com/bluesky-social/indigo/atproto/syntax" + "github.com/bluesky-social/indigo/repo" + "github.com/bluesky-social/indigo/xrpc" + "github.com/ipfs/go-cid" + "stream.place/streamplace/pkg/log" + "stream.place/streamplace/pkg/reposync" +) + +// placeStreamPrefix is the collection prefix shared by every streamplace +// lexicon. Because MST keys are "collection/rkey" sorted bytewise, all of those +// records live in one contiguous key range. +const placeStreamPrefix = "place.stream." + +// backfillRanges is the set of MST key ranges a backfill walks: everything +// under place.stream., plus one range per non-streamplace collection the +// firehose accepts. +// +// It is derived from CollectionFilter at runtime so the backfill and the +// firehose can never drift apart about which records this node indexes. +func backfillRanges() []reposync.KeyRange { + ranges := []reposync.KeyRange{reposync.PrefixRange(placeStreamPrefix)} + for _, nsid := range CollectionFilter { + if strings.HasPrefix(nsid, placeStreamPrefix) { + continue + } + // The trailing slash keeps "app.bsky.feed.post" from also matching + // "app.bsky.feed.postgate". + ranges = append(ranges, reposync.PrefixRange(nsid+"/")) + } + return ranges +} + +// backfillRepo indexes every record we care about from ident's repo. It returns +// the repo revision the index is now consistent with, and the MST root CID that +// revision committed to (empty if the fallback path was used, which never sees +// a verified root). +// +// The fast path walks only the subtrees holding records we index. Hosts that do +// not implement com.atproto.sync.getBlocks fall back to downloading the whole +// repo as a CAR. +func (atsync *ATProtoSynchronizer) backfillRepo(ctx context.Context, ident *identity.Identity, xrpcc *xrpc.Client) (string, string, error) { + rev, root, err := atsync.walkBackfill(ctx, ident, xrpcc) + if err == nil { + return rev, root, nil + } + if !isMethodNotSupported(err) { + // Anything else -- a bad signature, a malformed tree, a network + // failure -- must propagate. Falling back on a verification failure + // would make the verification decorative. + return "", "", err + } + log.Warn(ctx, "host does not support sync.getBlocks, falling back to full getRepo", + "pds", xrpcc.Host, "did", ident.DID.String(), "err", err) + rev, err = atsync.legacyBackfill(ctx, ident, xrpcc) + if err != nil { + return "", "", err + } + return rev, "", nil +} + +// walkBackfill does a verified, prefix-bounded walk of the remote repo, handing +// every record in range to the same indexing path the firehose uses. +func (atsync *ATProtoSynchronizer) walkBackfill(ctx context.Context, ident *identity.Identity, xrpcc *xrpc.Client) (string, string, error) { + did := ident.DID.String() + + dir := atsync.PLCDirectory + if dir == nil { + // resolveIdent initializes this lazily, and every caller goes through + // it first; be defensive rather than nil-panic. Note this is the + // *uncached* directory on purpose: a signing key cached from before a + // rotation would fail commit verification, and backfills are rare + // enough that the extra lookup does not matter. + dir = CustomDirectory(atsync.CLI.PLCURL) + } + + fetcher := &reposync.CachedFetcher{ + // Bounded lifetime: one cache per backfill, so the head fetch and the + // walk share blocks without holding a repo in memory afterwards. + Cache: reposync.NewMemoryBlockCache(), + Inner: &pdsLockedFetcher{ + lock: pdsLocks.GetLock(ident.PDSEndpoint()), + inner: &reposync.XRPCBlockFetcher{Client: xrpcc, DID: did}, + }, + } + + head, err := reposync.FetchVerifiedHead(ctx, xrpcc, fetcher, dir, did) + if err != nil { + return "", "", fmt.Errorf("failed to fetch verified head for %s from PDS %s: %w", did, xrpcc.Host, err) + } + + walker := &reposync.Walker{Fetcher: fetcher} + records := 0 + err = walker.WalkRanges(ctx, head.Root, backfillRanges(), func(path string, rcid cid.Cid, rec []byte) error { + nsid, rkey, err := syntax.ParseRepoPath(path) + if err != nil { + log.Warn(ctx, "failed to parse repo path", "k", path, "err", err) + return fmt.Errorf("could not parse repo path %s: %w", path, err) + } + log.Debug(ctx, "record type", "key", path, "type", nsid.String()) + + bs := rec + err = atsync.handleCreateUpdate(ctx, did, rkey, &bs, rcid.String(), nsid, false, true) + if err != nil { + log.Warn(ctx, "failed to handle create update", "err", err) + // invalid CBOR and stuff should get ignored, so we don't return + } + records++ + return nil + }) + if err != nil { + return "", "", fmt.Errorf("failed to walk repo for %s from PDS %s: %w", did, xrpcc.Host, err) + } + + log.Log(ctx, "walked repo", "did", did, "rev", head.Rev, "root", head.Root.String(), "records", records) + return head.Rev, head.Root.String(), nil +} + +// legacyBackfill is the pre-walker path: download the entire repo as a CAR and +// feed every record in it to the indexer. It is kept as the fallback for hosts +// without com.atproto.sync.getBlocks -- notably streamplace's own PDS, whose +// did:web repos are synced through here for VOD origin indexing. +func (atsync *ATProtoSynchronizer) legacyBackfill(ctx context.Context, ident *identity.Identity, xrpcc *xrpc.Client) (string, error) { + rev := "" + pdsLock := pdsLocks.GetLock(ident.PDSEndpoint()) + pdsLock.Lock() + repoBytes, err := SyncGetRepo(ctx, xrpcc, ident.DID.String(), rev) + pdsLock.Unlock() + if err != nil { + return "", fmt.Errorf("failed to fetch repo for %s from PDS %s: %w", ident.DID.String(), xrpcc.Host, err) + } + + log.Debug(ctx, "got diff", "bytes", len(repoBytes)) + + r, err := repo.ReadRepoFromCar(ctx, bytes.NewReader(repoBytes)) + if err != nil { + return "", fmt.Errorf("failed to parse repo CAR data for %s: %w", ident.DID.String(), err) + } + // extract DID from repo commit + sc := r.SignedCommit() + signerDID, err := syntax.ParseDID(sc.Did) + if err != nil { + return "", fmt.Errorf("invalid DID in repo commit for %s: %w", ident.DID.String(), err) + } + if signerDID != ident.DID { + return "", fmt.Errorf("signer DID %s does not match identity %s", signerDID, ident.DID.String()) + } + + err = r.ForEach(ctx, "", func(k string, v cid.Cid) error { + nsid, rkey, err := syntax.ParseRepoPath(k) + if err != nil { + log.Warn(ctx, "failed to parse repo path", "k", k, "err", err) + return fmt.Errorf("could not parse repo path %s: %w", k, err) + } + _, bs, err := r.GetRecordBytes(ctx, k) + if err != nil { + log.Warn(ctx, "failed to get record bytes", "k", k, "rkey", rkey, "err", err) + return fmt.Errorf("could not retrieve record bytes for %s (rkey: %s): %w", k, rkey, err) + } + log.Debug(ctx, "record type", "key", k, "type", nsid.String()) + + err = atsync.handleCreateUpdate(ctx, signerDID.String(), rkey, bs, v.String(), nsid, false, true) + if err != nil { + log.Warn(ctx, "failed to handle create update", "err", err) + // invalid CBOR and stuff should get ignored, so + // return fmt.Errorf("failed to process record update for %s (type: %s): %w", k, nsid.String(), err) + } + return nil + }) + if err != nil { + return "", fmt.Errorf("failed to iterate over repo: %w", err) + } + + return sc.Rev, nil +} + +// pdsLockedFetcher serializes block fetches per PDS, the way the legacy path +// serializes its one big getRepo download. +// +// The lock is held only across the network call and never across a visitor +// callback: handleCreateUpdate can synchronously start a sync of another repo, +// which may live on the same host, and pdsLocks are plain mutexes. +type pdsLockedFetcher struct { + lock *sync.Mutex + inner reposync.BlockFetcher +} + +var _ reposync.BlockFetcher = (*pdsLockedFetcher)(nil) + +func (f *pdsLockedFetcher) GetBlocks(ctx context.Context, cids []cid.Cid) (map[cid.Cid][]byte, error) { + f.lock.Lock() + defer f.lock.Unlock() + return f.inner.GetBlocks(ctx, cids) +} + +// isMethodNotSupported reports whether err means "this host does not implement +// that XRPC method", which is the only failure the backfill is allowed to +// answer by falling back to a full getRepo. +// +// A host that has never heard of the route answers 404 (or 405 for the wrong +// verb); one that knows the lexicon but has not implemented it answers 501 +// and/or the MethodNotImplemented error name. 401 is in the list because of +// streamplace's own PDS specifically: unregistered /xrpc/* methods land on its +// wildcard proxy handler, which needs an OAuth session and answers +// "oauth session not found" with a 401 to an anonymous sync request. Node to +// node repo sync (VOD origin indexing) runs through exactly that path, so +// without this the fallback would never fire for did:web streamplace repos. +// Errors are unwrapped because reposync wraps everything with %w. +// +// A false positive costs one wasted getRepo attempt, whose own error then +// propagates -- it can never turn a verification failure into a silent success, +// because verification failures are not HTTP errors. +func isMethodNotSupported(err error) bool { + if err == nil { + return false + } + var xe *xrpc.Error + if errors.As(err, &xe) { + switch xe.StatusCode { + case http.StatusUnauthorized, http.StatusNotFound, http.StatusMethodNotAllowed, http.StatusNotImplemented: + return true + } + } + var xrpcErr *xrpc.XRPCError + if errors.As(err, &xrpcErr) && xrpcErr.ErrStr == "MethodNotImplemented" { + return true + } + return false +} diff --git a/pkg/atproto/backfill_walk_test.go b/pkg/atproto/backfill_walk_test.go new file mode 100644 index 00000000..55a67d74 --- /dev/null +++ b/pkg/atproto/backfill_walk_test.go @@ -0,0 +1,413 @@ +package atproto + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "net/http/httputil" + "net/url" + "sort" + "strings" + "testing" + "time" + + "github.com/bluesky-social/indigo/util" + "github.com/bluesky-social/indigo/xrpc" + "github.com/ipfs/go-cid" + glex "github.com/streamplace/glex/runtime" + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/appbsky" + "stream.place/streamplace/pkg/aqhttp" + "stream.place/streamplace/pkg/bus" + "stream.place/streamplace/pkg/comatproto" + "stream.place/streamplace/pkg/config" + "stream.place/streamplace/pkg/devenv" + "stream.place/streamplace/pkg/model" + "stream.place/streamplace/pkg/placestream" + "stream.place/streamplace/pkg/reposync" + "stream.place/streamplace/pkg/statedb" +) + +// TestBackfillWalk drives SyncBlueskyRepo's walker path against the Bluesky +// reference PDS: real com.atproto.sync.getLatestCommit/getBlocks, real MST, +// real did:plc key resolution. The firehose is deliberately not started, so +// every record indexed here got there through the backfill. +func TestBackfillWalk(t *testing.T) { + dev := devenv.WithDevEnv(t) + ctx := context.Background() + atsync, mod := backfillTestSynchronizer(t, dev) + + user := dev.CreateAccount(t) + + // In range, via the place.stream. prefix. + profile := createBackfillRecord(t, user, "place.stream.chat.profile", "self", &placestream.ChatProfile{}) + msg1 := createBackfillRecord(t, user, "place.stream.chat.message", "", chatMessageRecord(user.DID, "hello one")) + msg2 := createBackfillRecord(t, user, "place.stream.chat.message", "", chatMessageRecord(user.DID, "hello two")) + // In range, via a CollectionFilter collection (sorts well below place.stream.). + displayName := "Backfill Tester" + bskyProfile := createBackfillRecord(t, user, "app.bsky.actor.profile", "self", &appbsky.ActorProfile{DisplayName: &displayName}) + // Out of range, and adjacent to the app.bsky.actor.profile/ range on the + // side where an off-by-one prefix bound would leak. + createBackfillRecord(t, user, "app.bsky.actor.status", "", rawBackfillRecord(t, "app.bsky.actor.status", map[string]any{ + "status": "app.bsky.actor.status#live", + "createdAt": time.Now().UTC().Format(util.ISO8601), + })) + // Out of range, above everything we index. + createBackfillRecord(t, user, "zzz.example.thing", "", rawBackfillRecord(t, "zzz.example.thing", map[string]any{ + "createdAt": time.Now().UTC().Format(util.ISO8601), + })) + + wantPaths := []string{bskyProfile, profile, msg1, msg2} + sort.Strings(wantPaths) + + // Walk the ranges the backfill uses, directly. This both waits for the PDS + // to have committed everything and proves the out-of-range records are + // never even fetched, which SyncBlueskyRepo alone cannot show (nothing + // indexes an app.bsky.actor.status either way). + var head *reposync.Head + err := untilNoErrors(t, func() error { + xrpcc := &xrpc.Client{Host: dev.PDSURL, Client: &aqhttp.Client} + fetcher := &reposync.CachedFetcher{ + Cache: reposync.NewMemoryBlockCache(), + Inner: &reposync.XRPCBlockFetcher{Client: xrpcc, DID: user.DID}, + } + h, err := reposync.FetchVerifiedHead(ctx, xrpcc, fetcher, dev.TestDirectory(), user.DID) + if err != nil { + return err + } + var got []string + walker := &reposync.Walker{Fetcher: fetcher} + err = walker.WalkRanges(ctx, h.Root, backfillRanges(), func(path string, rcid cid.Cid, rec []byte) error { + got = append(got, path) + return nil + }) + if err != nil { + return err + } + sort.Strings(got) + if strings.Join(got, ",") != strings.Join(wantPaths, ",") { + return fmt.Errorf("walked %v, want %v", got, wantPaths) + } + head = h + return nil + }) + require.NoError(t, err, "direct walk of backfill ranges") + + repo, err := atsync.SyncBlueskyRepoCached(ctx, user.DID) + require.NoError(t, err) + require.Equal(t, user.DID, repo.DID) + require.Equal(t, head.Rev, repo.Version, "repo row should record the rev it was synced to") + require.Equal(t, head.Root.String(), repo.RootCID, "repo row should record the verified MST root") + + // And it is durable, not just what SyncBlueskyRepo happened to return. + stored, err := mod.GetRepo(user.DID) + require.NoError(t, err) + require.Equal(t, head.Rev, stored.Version) + require.Equal(t, head.Root.String(), stored.RootCID) + + messages, err := mod.MostRecentChatMessages(user.DID) + require.NoError(t, err) + require.Len(t, messages, 2) + texts := []string{ + messages[0].Record.Val.(*placestream.ChatMessage).Text, + messages[1].Record.Val.(*placestream.ChatMessage).Text, + } + sort.Strings(texts) + require.Equal(t, []string{"hello one", "hello two"}, texts) + + chatProfile, err := mod.GetChatProfile(ctx, user.DID) + require.NoError(t, err) + require.NotNil(t, chatProfile, "place.stream.chat.profile should have been indexed") + + indexedBsky, err := mod.GetBskyProfile(ctx, user.DID, false) + require.NoError(t, err) + require.NotNil(t, indexedBsky, "app.bsky.actor.profile should have been indexed") + require.Equal(t, displayName, *indexedBsky.DisplayName) +} + +// TestBackfillWedgeHeals covers the placeholder-row semantics: a repo row with +// an empty Version is a backfill that never finished and must be retried, while +// a row with a Version is authoritative and must not cause any network traffic. +func TestBackfillWedgeHeals(t *testing.T) { + dev := devenv.WithDevEnv(t) + ctx := context.Background() + atsync, mod := backfillTestSynchronizer(t, dev) + + user := dev.CreateAccount(t) + createBackfillRecord(t, user, "place.stream.chat.message", "", chatMessageRecord(user.DID, "wedged")) + + // Exactly what a crashed backfill leaves behind. + require.NoError(t, mod.UpdateRepo(&model.Repo{ + DID: user.DID, + PDS: dev.PDSURL, + Handle: user.Handle, + Version: "", + })) + + err := untilNoErrors(t, func() error { + repo, err := atsync.SyncBlueskyRepoCached(ctx, user.DID) + if err != nil { + return err + } + if repo.Version == "" { + return fmt.Errorf("repo still has no version") + } + messages, err := mod.MostRecentChatMessages(user.DID) + if err != nil { + return err + } + if len(messages) != 1 { + return fmt.Errorf("expected 1 message, got %d", len(messages)) + } + return nil + }) + require.NoError(t, err, "an incomplete backfill should be re-run") + + stored, err := mod.GetRepo(user.DID) + require.NoError(t, err) + require.NotEmpty(t, stored.Version) + require.NotEmpty(t, stored.RootCID) + + // The inverse: a row with a Version short-circuits before anything is + // resolved or fetched. This DID exists nowhere, so any attempt to sync it + // would fail at identity resolution. + missingDID := "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa" + require.NoError(t, mod.UpdateRepo(&model.Repo{ + DID: missingDID, + PDS: "http://127.0.0.1:1", + Version: "3lbogus000000", + })) + got, err := atsync.SyncBlueskyRepoCached(ctx, missingDID) + require.NoError(t, err, "a complete repo row must be returned without touching the network") + require.Equal(t, "3lbogus000000", got.Version) + + // Same unresolvable DID, but wedged: now it must actually try to sync, and + // fail, rather than hand back the placeholder forever. + require.NoError(t, mod.UpdateRepo(&model.Repo{ + DID: missingDID, + PDS: "http://127.0.0.1:1", + Version: "", + })) + _, err = atsync.SyncBlueskyRepoCached(ctx, missingDID) + require.Error(t, err, "a placeholder row must not short-circuit the sync") +} + +// TestBackfillFallsBackToGetRepo puts a host in front of the dev PDS that +// serves everything except the two sync methods the walk needs -- which is +// exactly what streamplace's own PDS looks like to another node today -- and +// checks that the backfill quietly finishes over the legacy full-CAR path. +func TestBackfillFallsBackToGetRepo(t *testing.T) { + dev := devenv.WithDevEnv(t) + ctx := context.Background() + atsync, mod := backfillTestSynchronizer(t, dev) + + user := dev.CreateAccount(t) + createBackfillRecord(t, user, "place.stream.chat.message", "", chatMessageRecord(user.DID, "fallback")) + + target, err := url.Parse(dev.PDSURL) + require.NoError(t, err) + reverse := httputil.NewSingleHostReverseProxy(target) + proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasSuffix(r.URL.Path, "com.atproto.sync.getLatestCommit"), + strings.HasSuffix(r.URL.Path, "com.atproto.sync.getBlocks"): + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusNotFound) + _, _ = w.Write([]byte(`{"error":"MethodNotImplemented","message":"not registered"}`)) + default: + reverse.ServeHTTP(w, r) + } + })) + defer proxy.Close() + + // A finished repo row, so that indexing a chat message does not kick off a + // second sync of the same DID against the real (unproxied) PDS and index + // the record for us. + require.NoError(t, mod.UpdateRepo(&model.Repo{ + DID: user.DID, + PDS: proxy.URL, + Handle: user.Handle, + Version: "3lpretend0000", + })) + + ident, err := atsync.resolveIdent(ctx, user.DID, false) + require.NoError(t, err) + + var rev, root string + err = untilNoErrors(t, func() error { + var err error + rev, root, err = atsync.backfillRepo(ctx, ident, &xrpc.Client{Host: proxy.URL, Client: &aqhttp.Client}) + if err != nil { + return err + } + messages, err := mod.MostRecentChatMessages(user.DID) + if err != nil { + return err + } + if len(messages) != 1 { + return fmt.Errorf("expected 1 message, got %d", len(messages)) + } + return nil + }) + require.NoError(t, err, "backfill should have fallen back to getRepo") + require.NotEmpty(t, rev, "the legacy path still reports the commit rev") + require.Empty(t, root, "the legacy path has no verified MST root to record") +} + +func TestIsMethodNotSupported(t *testing.T) { + for _, tc := range []struct { + name string + err error + want bool + }{ + {"nil", nil, false}, + {"wrapped 404", fmt.Errorf("getBlocks: %w", &xrpc.Error{StatusCode: http.StatusNotFound}), true}, + {"wrapped 405", fmt.Errorf("getBlocks: %w", &xrpc.Error{StatusCode: http.StatusMethodNotAllowed}), true}, + { + // What streamplace's own PDS answers for an unregistered + // /xrpc/* method: the wildcard proxy wants an OAuth session. + "wrapped 401 from the spxrpc wildcard proxy", + fmt.Errorf("getLatestCommit: %w", &xrpc.Error{ + StatusCode: http.StatusUnauthorized, + Wrapped: &xrpc.XRPCError{Message: "oauth session not found"}, + }), + true, + }, + {"wrapped 501", fmt.Errorf("getBlocks: %w", &xrpc.Error{StatusCode: http.StatusNotImplemented}), true}, + { + "doubly wrapped 404 with body", + fmt.Errorf("walking: %w", fmt.Errorf("getBlocks: %w", &xrpc.Error{ + StatusCode: http.StatusNotFound, + Wrapped: &xrpc.XRPCError{ErrStr: "NotFound", Message: "no such route"}, + })), + true, + }, + { + "MethodNotImplemented error name", + fmt.Errorf("getLatestCommit: %w", &xrpc.XRPCError{ErrStr: "MethodNotImplemented", Message: "nope"}), + true, + }, + { + "400 InvalidRequest", + fmt.Errorf("getBlocks: %w", &xrpc.Error{ + StatusCode: http.StatusBadRequest, + Wrapped: &xrpc.XRPCError{ErrStr: "InvalidRequest", Message: "cids/0 must be a cid string"}, + }), + false, + }, + { + "400 RepoNotFound", + fmt.Errorf("getLatestCommit: %w", &xrpc.Error{ + StatusCode: http.StatusBadRequest, + Wrapped: &xrpc.XRPCError{ErrStr: "RepoNotFound", Message: "could not find repo"}, + }), + false, + }, + {"block mismatch", fmt.Errorf("fetching: %w", reposync.ErrBlockMismatch), false}, + {"missing block", fmt.Errorf("fetching: %w", reposync.ErrMissingBlock), false}, + {"canceled", fmt.Errorf("walking: %w", context.Canceled), false}, + {"plain error", errors.New("connection refused"), false}, + } { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.want, isMethodNotSupported(tc.err)) + }) + } +} + +func TestBackfillRanges(t *testing.T) { + ranges := backfillRanges() + // One for place.stream., plus one per non-streamplace collection the + // firehose accepts. + want := 1 + for _, nsid := range CollectionFilter { + if !strings.HasPrefix(nsid, placeStreamPrefix) { + want++ + } + } + require.Len(t, ranges, want) + + inRange := func(key string) bool { + for _, r := range ranges { + if r.Lo != nil && key < string(r.Lo) { + continue + } + if r.Hi != nil && key >= string(r.Hi) { + continue + } + return true + } + return false + } + require.True(t, inRange("place.stream.chat.message/3l")) + require.True(t, inRange("place.stream.live.recommendations/self")) + require.True(t, inRange("app.bsky.actor.profile/self")) + require.True(t, inRange("app.bsky.feed.post/3l")) + require.False(t, inRange("app.bsky.actor.status/3l")) + require.False(t, inRange("app.bsky.feed.postgate/3l")) + require.False(t, inRange("app.bsky.feed.like/3l")) + require.False(t, inRange("place.strea.thing/3l")) + require.False(t, inRange("place.streamx.thing/3l")) + require.False(t, inRange("zzz.example.thing/3l")) +} + +func backfillTestSynchronizer(t *testing.T, dev *devenv.DevEnv) (*ATProtoSynchronizer, model.Model) { + t.Helper() + cli := config.CLI{ + BroadcasterHost: "example.com", + DBURL: ":memory:", + RelayHost: strings.ReplaceAll(dev.PDSURL, "http://", "ws://"), + PLCURL: dev.PLCURL, + DataDir: t.TempDir(), + } + mod, err := model.MakeDB(":memory:") + require.NoError(t, err) + state, err := statedb.MakeDB(context.Background(), &cli, nil, mod) + require.NoError(t, err) + return &ATProtoSynchronizer{ + CLI: &cli, + StatefulDB: state, + Model: mod, + Bus: bus.NewBus(), + }, mod +} + +func chatMessageRecord(streamerDID, text string) *placestream.ChatMessage { + return &placestream.ChatMessage{ + LexiconTypeID: "place.stream.chat.message", + Text: text, + CreatedAt: time.Now().UTC().Format(util.ISO8601), + Streamer: streamerDID, + } +} + +// rawBackfillRecord builds a record for a lexicon nothing in this repo +// generates code for. The reference PDS stores unknown collections verbatim. +func rawBackfillRecord(t *testing.T, typ string, fields map[string]any) glex.Record { + t.Helper() + m := map[string]any{"$type": typ} + for k, v := range fields { + m[k] = v + } + rec, err := glex.RawJSON(m) + require.NoError(t, err) + return rec +} + +// createBackfillRecord writes one record and returns its MST key +// ("collection/rkey"). rkey may be empty to let the PDS mint a TID. +func createBackfillRecord(t *testing.T, acct *devenv.DevEnvAccount, collection, rkey string, rec glex.Record) string { + t.Helper() + in := &comatproto.RepoCreateRecord_Input{ + Collection: collection, + Repo: acct.DID, + Record: &glex.LexiconTypeDecoder{Val: rec}, + } + if rkey != "" { + in.Rkey = &rkey + } + out, err := comatproto.RepoCreateRecord(context.Background(), acct.XRPC, in) + require.NoError(t, err, "creating %s record", collection) + return collection + "/" + out.Uri[strings.LastIndex(out.Uri, "/")+1:] +}