diff --git a/appview/sitefeed/feed.go b/appview/sitefeed/feed.go index 9b563fd13..6b0a722a0 100644 --- a/appview/sitefeed/feed.go +++ b/appview/sitefeed/feed.go @@ -6,7 +6,6 @@ import ( "errors" "fmt" "log/slog" - "strconv" "sync" "time" @@ -141,43 +140,43 @@ func (f *Feed) consumer(host string, noTLS bool) *knotfeed.Consumer { LoadCursor: f.loadCursor(host), StoreCursor: f.storeCursor(host), Handle: f.handle(host), - OutdatedReplay: func(context.Context) int64 { - f.logger.Warn("site feed cursor is behind the knot, resuming live", "host", host) - return 0 + OutdatedReplay: func(_ context.Context, feed knotfeed.Feed) knotfeed.Cursor { + f.logger.Warn("site feed cursor is behind the knot, resuming live", "host", host, "feed", feed) + return feed.Live(time.Now()) }, } } -func (f *Feed) loadCursor(host string) func(context.Context) (int64, error) { - return func(ctx context.Context) (int64, error) { +func (f *Feed) loadCursor(host string) func(context.Context) (knotfeed.Cursor, error) { + return func(ctx context.Context) (knotfeed.Cursor, error) { if f.rdb == nil { - return 0, nil + return knotfeed.Cursor{}, nil } raw, err := f.rdb.Get(ctx, f.cursorKey(host)).Result() if errors.Is(err, redis.Nil) { - return 0, nil + return knotfeed.Cursor{}, nil } if err != nil { - return 0, fmt.Errorf("loading the site feed cursor: %w", err) + return knotfeed.Cursor{}, fmt.Errorf("loading site feed cursor: %w", err) } - seq, err := strconv.ParseInt(raw, 10, 64) + cursor, err := knotfeed.ParseCursor(raw) if err != nil { f.logger.Warn("unreadable site feed cursor, resuming live", "host", host, "err", err) if err := f.rdb.Del(ctx, f.cursorKey(host)).Err(); err != nil { - return 0, fmt.Errorf("clearing an unreadable site feed cursor: %w", err) + return knotfeed.Cursor{}, fmt.Errorf("clearing unreadable site feed cursor: %w", err) } - return 0, nil + return knotfeed.Cursor{}, nil } - return seq, nil + return cursor, nil } } -func (f *Feed) storeCursor(host string) func(context.Context, int64) error { - return func(ctx context.Context, seq int64) error { +func (f *Feed) storeCursor(host string) func(context.Context, knotfeed.Cursor) error { + return func(ctx context.Context, cursor knotfeed.Cursor) error { if f.rdb == nil { return nil } - return f.rdb.Set(ctx, f.cursorKey(host), seq, 0).Err() + return f.rdb.Set(ctx, f.cursorKey(host), cursor.Encode(), 0).Err() } } @@ -201,7 +200,7 @@ func (f *Feed) handle(host string) func(context.Context, knotfeed.Message) error } } -func (f *Feed) refOp(ctx context.Context, host, repoDid string, op knotfeed.RecordOp) error { +func (f *Feed) refOp(ctx context.Context, host string, repoDid syntax.DID, op knotfeed.RecordOp) error { logger := f.logger.With("knot", host, "repo_did", repoDid) refname, ok := knotfeed.UnescapeRkey(op.Rkey) @@ -220,7 +219,7 @@ func (f *Feed) refOp(ctx context.Context, host, repoDid string, op knotfeed.Reco return nil } - repo, err := db.GetRepoByDid(f.d, repoDid) + repo, err := db.GetRepoByDid(f.d, repoDid.String()) if errors.Is(err, sql.ErrNoRows) { return nil } @@ -244,7 +243,7 @@ func (f *Feed) refOp(ctx context.Context, host, repoDid string, op knotfeed.Reco return nil } -func (f *Feed) maybeDeploy(ctx context.Context, repo *models.Repo, branch, sha string) error { +func (f *Feed) maybeDeploy(ctx context.Context, repo *models.Repo, branch string, sha knotfeed.ObjectID) error { if f.cf == nil || !f.cf.Enabled() { return nil } @@ -263,7 +262,7 @@ func (f *Feed) maybeDeploy(ctx context.Context, repo *models.Repo, branch, sha s case f.deploySlots <- struct{}{}: default: f.logger.Warn("deploy queue saturated, dropping the deploy", "repo", repo.RepoDid, "branch", branch) - f.recordDroppedDeploy(repo, siteConfig, sha, "deploy queue saturated") + f.recordDroppedDeploy(repo, siteConfig, sha.String(), "deploy queue saturated") return nil } go func() { @@ -271,11 +270,11 @@ func (f *Feed) maybeDeploy(ctx context.Context, repo *models.Repo, branch, sha s mu, _ := f.deployLocks.LoadOrStore(repo.RepoDid, &sync.Mutex{}) mu.(*sync.Mutex).Lock() defer mu.(*sync.Mutex).Unlock() - if latest, ok := f.pendingSha.Load(repo.RepoDid); !ok || latest.(string) != sha { + if latest, ok := f.pendingSha.Load(repo.RepoDid); !ok || latest.(knotfeed.ObjectID) != sha { f.logger.Info("superseded push deploy skipped", "repo", repo.RepoDid, "branch", branch, "sha", sha) return } - f.triggerDeploy(context.WithoutCancel(ctx), repo, siteConfig, sha) + f.triggerDeploy(context.WithoutCancel(ctx), repo, siteConfig, sha.String()) }() return nil } diff --git a/appview/sitefeed/feed_test.go b/appview/sitefeed/feed_test.go index 158ec5d6d..f6b197c77 100644 --- a/appview/sitefeed/feed_test.go +++ b/appview/sitefeed/feed_test.go @@ -14,6 +14,7 @@ import ( "tangled.org/core/appview/db" "tangled.org/core/appview/models" "tangled.org/core/knotfeed" + "tangled.org/core/tapc" "tangled.org/core/log" cbg "github.com/whyrusleeping/cbor-gen" @@ -24,6 +25,8 @@ const ( feedTestOtherHost = "other.knot.invalid" feedTestRepoDid = "did:plc:limpet" feedTestOwner = "did:plc:akshay" + feedTestSha = "bcfa6359b037bc411e9626260cc3dd9a28408d2c" + feedTestOtherSha = "9eab6c4f29bc9e609ec52888cd6a4ef716b567c9" ) var errBoom = errors.New("boom") @@ -121,7 +124,7 @@ func encodeRefRecord(t *testing.T, sha string) []byte { return out.Bytes() } -func refOpFor(t *testing.T, refname, action, sha string) knotfeed.RecordOp { +func refOpFor(t *testing.T, refname string, action tapc.RecordAction, sha string) knotfeed.RecordOp { t.Helper() rkey, _ := knotfeed.EscapeRefname(refname) op := knotfeed.RecordOp{Action: action, Collection: knotfeed.GitRefCollection, Rkey: rkey} @@ -131,6 +134,15 @@ func refOpFor(t *testing.T, refname, action, sha string) knotfeed.RecordOp { return op } +func object(t *testing.T, hex string) knotfeed.ObjectID { + t.Helper() + sha, ok := knotfeed.ParseObjectID(hex) + if !ok { + t.Fatalf("ParseObjectID(%q)", hex) + } + return sha +} + func pollFor(t *testing.T, what string, ready func() bool) { t.Helper() deadline := time.Now().Add(2 * time.Second) @@ -161,8 +173,7 @@ func TestFeedRefOp_BranchMatchTriggersDeploy(t *testing.T) { r := &deployRecorder{} f := feedFor(t, d, r.hook(nil)) - f.refOp(context.Background(), feedTestHost, feedTestRepoDid, refOpFor(t, "refs/heads/main", "create", "def456")) - + f.refOp(context.Background(), feedTestHost, feedTestRepoDid, refOpFor(t, "refs/heads/main", tapc.RecordCreateAction, feedTestSha)) pollFor(t, "the deploy call", func() bool { return len(r.got()) == 1 }) calls := r.got() if calls[0] != (deployCall{repoDid: feedTestRepoDid, branch: "main", dir: "/"}) { @@ -184,7 +195,7 @@ func TestFeedRefOp_BranchMatchTriggersDeploy(t *testing.T) { if deploy.Trigger != models.SiteDeployTriggerPush { t.Errorf("trigger = %q, want push", deploy.Trigger) } - if deploy.CommitSHA != "def456" || deploy.Branch != "main" || deploy.Dir != "/" { + if deploy.CommitSHA != feedTestSha || deploy.Branch != "main" || deploy.Dir != "/" { t.Errorf("deploy row = %+v, want the pushed sha, branch and dir", deploy) } } @@ -203,55 +214,55 @@ func TestFeedSkipsNonDeployingOps(t *testing.T) { name: "push to a branch other than the site branch", seed: seedSiteRepo, op: func(_ *testing.T, f *Feed) { - f.maybeDeploy(context.Background(), &models.Repo{RepoDid: feedTestRepoDid}, "other", "abc123") + f.maybeDeploy(context.Background(), &models.Repo{RepoDid: feedTestRepoDid}, "other", object(t, feedTestSha)) }, }, { name: "push without a site config", seed: seedFeedRepo, op: func(_ *testing.T, f *Feed) { - f.maybeDeploy(context.Background(), &models.Repo{RepoDid: feedTestRepoDid}, "main", "abc123") + f.maybeDeploy(context.Background(), &models.Repo{RepoDid: feedTestRepoDid}, "main", object(t, feedTestSha)) }, }, { name: "tag push", seed: seedSiteRepo, op: func(t *testing.T, f *Feed) { - f.refOp(context.Background(), feedTestHost, feedTestRepoDid, refOpFor(t, "refs/tags/v1", "create", "def456")) + f.refOp(context.Background(), feedTestHost, feedTestRepoDid, refOpFor(t, "refs/tags/v1", tapc.RecordCreateAction, feedTestSha)) }, }, { name: "deleted ref", seed: seedSiteRepo, op: func(t *testing.T, f *Feed) { - f.refOp(context.Background(), feedTestHost, feedTestRepoDid, refOpFor(t, "refs/heads/main", "delete", "")) + f.refOp(context.Background(), feedTestHost, feedTestRepoDid, refOpFor(t, "refs/heads/main", tapc.RecordDeleteAction, "")) }, }, { name: "record from a knot that doesn't host the repo", seed: seedSiteRepo, op: func(t *testing.T, f *Feed) { - f.refOp(context.Background(), feedTestOtherHost, feedTestRepoDid, refOpFor(t, "refs/heads/main", "create", "def456")) + f.refOp(context.Background(), feedTestOtherHost, feedTestRepoDid, refOpFor(t, "refs/heads/main", tapc.RecordCreateAction, feedTestSha)) }, }, { name: "repo missing from the index", op: func(t *testing.T, f *Feed) { - f.refOp(context.Background(), feedTestHost, "did:plc:unknown", refOpFor(t, "refs/heads/main", "create", "def456")) + f.refOp(context.Background(), feedTestHost, "did:plc:unknown", refOpFor(t, "refs/heads/main", tapc.RecordCreateAction, feedTestSha)) }, }, { name: "garbage rkey", seed: seedSiteRepo, op: func(_ *testing.T, f *Feed) { - f.refOp(context.Background(), feedTestHost, feedTestRepoDid, knotfeed.RecordOp{Action: "create", Collection: knotfeed.GitRefCollection, Rkey: "~zz", Bytes: []byte{0xff}}) + f.refOp(context.Background(), feedTestHost, feedTestRepoDid, knotfeed.RecordOp{Action: tapc.RecordCreateAction, Collection: knotfeed.GitRefCollection, Rkey: "~zz", Bytes: []byte{0xff}}) }, }, { name: "garbage record bytes", seed: seedSiteRepo, op: func(_ *testing.T, f *Feed) { - f.refOp(context.Background(), feedTestHost, feedTestRepoDid, knotfeed.RecordOp{Action: "create", Collection: knotfeed.GitRefCollection, Rkey: "refs~2fheads~2fmain", Bytes: []byte{0xff}}) + f.refOp(context.Background(), feedTestHost, feedTestRepoDid, knotfeed.RecordOp{Action: tapc.RecordCreateAction, Collection: knotfeed.GitRefCollection, Rkey: "refs~2fheads~2fmain", Bytes: []byte{0xff}}) }, }, { @@ -262,7 +273,7 @@ func TestFeedSkipsNonDeployingOps(t *testing.T) { Repo: feedTestRepoDid, Seq: 7, Records: []knotfeed.RecordOp{ - {Action: "create", Collection: "sh.tangled.repo", Rkey: "abc", Bytes: []byte("x")}, + {Action: tapc.RecordCreateAction, Collection: "sh.tangled.repo", Rkey: "abc", Bytes: []byte("x")}, }, }} if err := f.handle(feedTestHost)(context.Background(), msg); err != nil { @@ -293,7 +304,7 @@ func TestFeedTrigger_RecordsDeployFailure(t *testing.T) { r := &deployRecorder{} f := feedFor(t, d, r.hook(errBoom)) - f.maybeDeploy(context.Background(), &models.Repo{RepoDid: feedTestRepoDid, Knot: feedTestHost}, "main", "abc123") + f.maybeDeploy(context.Background(), &models.Repo{RepoDid: feedTestRepoDid, Knot: feedTestHost}, "main", object(t, feedTestSha)) pollFor(t, "the deploy call", func() bool { return len(r.got()) == 1 }) var deploy *models.SiteDeploy @@ -328,7 +339,7 @@ func TestFeedHandleFailsTheCommitWhenTheDatabaseIsDown(t *testing.T) { Repo: feedTestRepoDid, Seq: 9, Records: []knotfeed.RecordOp{ - refOpFor(t, "refs/heads/main", "create", "def456"), + refOpFor(t, "refs/heads/main", tapc.RecordCreateAction, feedTestSha), }, }} if err := f.handle(feedTestHost)(context.Background(), msg); err == nil { @@ -380,11 +391,11 @@ func TestFeedTrigger_LastPushWinsOverARacingDeploy(t *testing.T) { f := feedFor(t, d, r.hook(nil)) repo := &models.Repo{RepoDid: feedTestRepoDid, Knot: feedTestHost} - f.maybeDeploy(context.Background(), repo, "main", "shaA") - f.maybeDeploy(context.Background(), repo, "main", "shaB") + f.maybeDeploy(context.Background(), repo, "main", object(t, feedTestSha)) + f.maybeDeploy(context.Background(), repo, "main", object(t, feedTestOtherSha)) var winner models.SiteDeploy - pollFor(t, "a successful shaB deploy row", func() bool { + pollFor(t, "a successful deploy row for the later push", func() bool { rows, err := db.GetSiteDeploys(d, feedTestRepoDid, 10) if err != nil { t.Fatalf("GetSiteDeploys: %v", err) @@ -394,6 +405,6 @@ func TestFeedTrigger_LastPushWinsOverARacingDeploy(t *testing.T) { winner = row } } - return winner.CommitSHA == "shaB" && winner.Status == models.SiteDeployStatusSuccess + return winner.CommitSHA == feedTestOtherSha && winner.Status == models.SiteDeployStatusSuccess }) } diff --git a/knot2/interop/knotfeed_consumer_test.go b/knot2/interop/knotfeed_consumer_test.go index 3593d7054..823ef92cc 100644 --- a/knot2/interop/knotfeed_consumer_test.go +++ b/knot2/interop/knotfeed_consumer_test.go @@ -11,14 +11,16 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/bluesky-social/indigo/atproto/syntax" "tangled.org/core/knotfeed" + "tangled.org/core/tapc" ) func TestKnotfeedConsumerReadsRefRecords(t *testing.T) { addr := os.Getenv("KNOT_TEST_ADDR") require.NotEmpty(t, addr, "KNOT_TEST_ADDR names the knot's host:port") - repoDid := os.Getenv("KNOT_TEST_REPO_DID") - require.NotEmpty(t, repoDid, "KNOT_TEST_REPO_DID names the pushed repository") + repoDid, err := syntax.ParseDID(os.Getenv("KNOT_TEST_REPO_DID")) + require.NoError(t, err, "KNOT_TEST_REPO_DID names the pushed repository") editor := os.Getenv("KNOT_TEST_EDITOR") require.NotEmpty(t, editor, "KNOT_TEST_EDITOR names the pushing account") wantSha := os.Getenv("KNOT_TEST_SHA") @@ -34,11 +36,11 @@ func TestKnotfeedConsumerReadsRefRecords(t *testing.T) { NoTLS: true, ReplayFromStart: true, Logger: slog.Default(), - LoadCursor: func(context.Context) (int64, error) { - return cursor.Load(), nil + LoadCursor: func(context.Context) (knotfeed.Cursor, error) { + return knotfeed.NewCursor(knotfeed.FeedAtproto, cursor.Load()), nil }, - StoreCursor: func(_ context.Context, seq int64) error { - cursor.Store(seq) + StoreCursor: func(_ context.Context, cur knotfeed.Cursor) error { + cursor.Store(cur.Seq()) return nil }, Handle: func(_ context.Context, message knotfeed.Message) error { @@ -49,7 +51,7 @@ func TestKnotfeedConsumerReadsRefRecords(t *testing.T) { return nil } for _, op := range message.Commit.Records { - if op.Collection == knotfeed.GitRefCollection && op.Action == "create" { + if op.Collection == knotfeed.GitRefCollection && op.Action == tapc.RecordCreateAction { select { case records <- op: default: @@ -68,8 +70,8 @@ func TestKnotfeedConsumerReadsRefRecords(t *testing.T) { assert.Equal(t, "refs/heads/main", refname) rec, err := knotfeed.DecodeRefRecord(op.Bytes) require.NoError(t, err, "the frozen record envelope decodes") - assert.Equal(t, wantSha, rec.Sha, "the record carries the pushed tip") - assert.Equal(t, editor, rec.Editor, "the record names the pusher") + assert.Equal(t, wantSha, rec.Sha.String(), "the record carries the pushed tip") + assert.Equal(t, editor, rec.Editor.String(), "the record names the pusher") require.Len(t, rec.PushOptions, 1) assert.Equal(t, "skip-ci", rec.PushOptions[0], "the push options ride the undeclared field") require.Eventually(t, func() bool { return cursor.Load() >= int64(1) }, diff --git a/knotfeed/subscribe.go b/knotfeed/subscribe.go index 7ce514d90..75f5e44ca 100644 --- a/knotfeed/subscribe.go +++ b/knotfeed/subscribe.go @@ -7,6 +7,9 @@ import ( "fmt" "log/slog" "net/url" + "slices" + "strconv" + "strings" "time" "github.com/gorilla/websocket" @@ -19,21 +22,114 @@ const ( maxBackoff = 60 * time.Second maxFrameBytes = 8 << 20 subscribeReposNS = "com.atproto.sync.subscribeRepos" + eventsPath = "/events" defaultMaxHandlerAttempts = 3 defaultPoisonGrace = 10 * time.Minute ) +type Feed struct { + kind uint8 +} + +const ( + atprotoKind uint8 = iota + legacyKind +) + +var ( + FeedAtproto = Feed{kind: atprotoKind} + FeedLegacy = Feed{kind: legacyKind} +) + +type feedSpec struct { + token string + name string + path string + msgKind int + other Feed + decode func([]byte, *slog.Logger) (Message, error) +} + +var feedSpecs = [...]feedSpec{ + atprotoKind: { + token: "atproto", name: "atproto firehose", path: "/xrpc/" + subscribeReposNS, + msgKind: websocket.BinaryMessage, other: FeedLegacy, decode: Decode, + }, + legacyKind: { + token: "legacy", name: "knot event stream", path: eventsPath, + msgKind: websocket.TextMessage, other: FeedAtproto, decode: decodeLegacy, + }, +} + +func (f Feed) spec() feedSpec { return feedSpecs[f.kind] } + +func (f Feed) Token() string { return f.spec().token } + +func (f Feed) String() string { return f.spec().name } + +func (f Feed) other() Feed { return f.spec().other } + +func (f Feed) Live(now time.Time) Cursor { + if f == FeedLegacy { + return NewCursor(f, now.UnixNano()) + } + return NewCursor(f, 0) +} + +var ErrUnrecognizedFeed = errors.New("unrecognized feed token") + +func ParseFeed(token string) (Feed, error) { + kind := slices.IndexFunc(feedSpecs[:], func(spec feedSpec) bool { return spec.token == token }) + if kind < 0 { + return FeedAtproto, fmt.Errorf("%w: %q", ErrUnrecognizedFeed, token) + } + return Feed{kind: uint8(kind)}, nil +} + +type Cursor struct { + feed Feed + seq int64 +} + +func NewCursor(feed Feed, seq int64) Cursor { + return Cursor{feed: feed, seq: seq} +} + +func (c Cursor) Feed() Feed { return c.feed } + +func (c Cursor) Seq() int64 { return c.seq } + +func (c Cursor) Encode() string { + return c.feed.Token() + ":" + strconv.FormatInt(c.seq, 10) +} + +func ParseCursor(raw string) (Cursor, error) { + token, rest, named := strings.Cut(raw, ":") + if !named { + token, rest = FeedAtproto.Token(), raw + } + feed, err := ParseFeed(token) + if err != nil { + return Cursor{}, err + } + seq, err := strconv.ParseInt(rest, 10, 64) + if err != nil { + return Cursor{}, fmt.Errorf("cursor %q: %w", raw, err) + } + return NewCursor(feed, seq), nil +} + type Consumer struct { Host string NoTLS bool Dialer *websocket.Dialer Logger *slog.Logger - LoadCursor func(context.Context) (int64, error) - StoreCursor func(context.Context, int64) error + LoadCursor func(context.Context) (Cursor, error) + StoreCursor func(context.Context, Cursor) error Handle func(context.Context, Message) error - OutdatedReplay func(context.Context) int64 + OutdatedReplay func(context.Context, Feed) Cursor OnConnectError func(error) ReplayFromStart bool @@ -43,6 +139,9 @@ type Consumer struct { now func() time.Time + current Feed + adopted bool + proven bool poisonSeq int64 poisonAttempts int poisonSince time.Time @@ -50,7 +149,6 @@ type Consumer struct { func (c *Consumer) Run(ctx context.Context) error { c.fill() - logger := c.Logger backoff := initialBackoff for { if ctx.Err() != nil { @@ -59,10 +157,7 @@ func (c *Consumer) Run(ctx context.Context) error { advanced, resync, err := c.session(ctx) switch { case err != nil: - if c.OnConnectError != nil && isConnectError(err) { - c.OnConnectError(err) - } - logger.Error("firehose session failed", "host", c.Host, "err", err) + c.reactTo(err) case resync: advanced = true } @@ -84,6 +179,29 @@ func (c *Consumer) Run(ctx context.Context) error { } } +func (c *Consumer) reactTo(err error) { + var mismatch feedMismatch + switch { + case errors.As(err, &mismatch): + case isConnectError(err): + if c.OnConnectError != nil { + c.OnConnectError(err) + } + if c.proven { + c.Logger.Error("knot turned us away on a feed it had already served", + "host", c.Host, "feed", c.current, "err", err) + return + } + default: + c.Logger.Error("session died for its own reasons", + "host", c.Host, "feed", c.current, "err", err) + return + } + c.Logger.Warn("trying the other feed, since this knot won't serve the one we asked for", + "host", c.Host, "tried", c.current, "next", c.current.other(), "err", err) + c.current, c.proven = c.current.other(), false +} + type connectError struct{ err error } func (e connectError) Error() string { return e.err.Error() } @@ -95,35 +213,53 @@ func isConnectError(err error) bool { return errors.As(err, &ce) } +type feedMismatch struct{ feed Feed } + +func (e feedMismatch) Error() string { + return fmt.Sprintf("%s sent the %s's frames", e.feed, e.feed.other()) +} + func (c *Consumer) session(ctx context.Context) (advanced bool, resync bool, err error) { - cursor, err := c.LoadCursor(ctx) + stored, err := c.LoadCursor(ctx) if err != nil { return false, false, fmt.Errorf("loading cursor: %w", err) } + if !c.adopted { + c.adopted = true + if stored.Seq() > 0 { + c.current = stored.Feed() + } + } + cursor := c.sessionCursor(stored) conn, _, err := c.Dialer.DialContext(ctx, c.url(cursor), nil) if err != nil { return false, false, connectError{err} } + c.proven = true defer conn.Close() watcher := context.AfterFunc(ctx, func() { conn.Close() }) defer watcher() conn.SetReadLimit(maxFrameBytes) - c.Logger.Info("subscribed to the firehose", "host", c.Host, "cursor", cursor) + c.Logger.Info("subscribed to knot feed", "host", c.Host, "feed", c.current, "cursor", cursor) conn.SetPingHandler(func(payload string) error { conn.SetReadDeadline(time.Now().Add(livenessTimeout)) return conn.WriteControl(websocket.PongMessage, []byte(payload), time.Now().Add(pongWriteTimeout)) }) conn.SetReadDeadline(time.Now().Add(livenessTimeout)) + want := c.current.spec().msgKind for { - _, data, err := conn.ReadMessage() + kind, data, err := conn.ReadMessage() if err != nil { return advanced, false, err } conn.SetReadDeadline(time.Now().Add(livenessTimeout)) - message, err := Decode(data, c.Logger) + if kind != want { + return advanced, false, feedMismatch{feed: c.current} + } + message, err := c.decode(data) if err != nil { - c.Logger.Error("undecodable firehose frame", "host", c.Host, "err", err) + c.Logger.Error("undecodable knot feed frame", "host", c.Host, "feed", c.current, "err", err) continue } resync, err := c.dispatch(ctx, message, &cursor) @@ -139,19 +275,21 @@ func (c *Consumer) session(ctx context.Context) (advanced bool, resync bool, err } } -func (c *Consumer) dispatch(ctx context.Context, message Message, cursor *int64) (bool, error) { +func (c *Consumer) decode(data []byte) (Message, error) { + message, err := c.current.spec().decode(data, c.Logger) + message.Feed = c.current + return message, err +} + +func (c *Consumer) dispatch(ctx context.Context, message Message, cursor *Cursor) (bool, error) { if message.Type == TypeCommit { return c.commit(ctx, message, cursor) } if c.unreachable(message) { c.Logger.Warn("firehose can't serve our cursor, resuming live", "host", c.Host, "error", c.resumeReason(message)) - replay := int64(0) - if c.OutdatedReplay != nil { - replay = c.OutdatedReplay(ctx) - } - *cursor = replay - if err := c.StoreCursor(ctx, replay); err != nil { - return false, fmt.Errorf("storing cursor: %w", err) + *cursor = c.replayCursor(ctx) + if err := c.store(ctx, *cursor); err != nil { + return false, err } return true, nil } @@ -163,7 +301,21 @@ func (c *Consumer) dispatch(ctx context.Context, message Message, cursor *int64) return false, nil } -func (c *Consumer) commit(ctx context.Context, message Message, cursor *int64) (bool, error) { +func (c *Consumer) replayCursor(ctx context.Context) Cursor { + live := c.current.Live(c.now()) + if c.OutdatedReplay == nil { + return live + } + replay := c.OutdatedReplay(ctx, c.current) + if replay.Feed() != c.current { + c.Logger.Warn("replay hook named the other feed, so resuming live instead", + "host", c.Host, "feed", c.current, "replay", replay.Feed()) + return live + } + return replay +} + +func (c *Consumer) commit(ctx context.Context, message Message, cursor *Cursor) (bool, error) { if c.Handle != nil { if err := c.Handle(ctx, message); err != nil { if !c.poison(message.Commit.Seq) { @@ -173,13 +325,20 @@ func (c *Consumer) commit(ctx context.Context, message Message, cursor *int64) ( "host", c.Host, "seq", message.Commit.Seq, "attempts", c.poisonAttempts, "err", err) } } - *cursor = message.Commit.Seq - if err := c.StoreCursor(ctx, *cursor); err != nil { - return false, fmt.Errorf("storing cursor: %w", err) + *cursor = NewCursor(c.current, message.Commit.Seq) + if err := c.store(ctx, *cursor); err != nil { + return false, err } return false, nil } +func (c *Consumer) store(ctx context.Context, cursor Cursor) error { + if err := c.StoreCursor(ctx, cursor); err != nil { + return fmt.Errorf("storing cursor: %w", err) + } + return nil +} + func (c *Consumer) poison(seq int64) bool { if seq != c.poisonSeq { c.poisonSeq = seq @@ -204,18 +363,29 @@ func (c *Consumer) resumeReason(message Message) string { return message.InfoName } -func (c *Consumer) url(cursor int64) string { +func (c *Consumer) url(cursor Cursor) string { scheme := "wss" if c.NoTLS { scheme = "ws" } - endpoint := url.URL{Scheme: scheme, Host: c.Host, Path: "/xrpc/" + subscribeReposNS} - if cursor > 0 || (cursor == 0 && c.ReplayFromStart) { - endpoint.RawQuery = "cursor=" + fmt.Sprint(cursor) + endpoint := url.URL{Scheme: scheme, Host: c.Host, Path: c.current.spec().path} + if seq := cursor.Seq(); seq > 0 || (seq == 0 && c.ReplayFromStart) { + endpoint.RawQuery = "cursor=" + strconv.FormatInt(seq, 10) } return endpoint.String() } +func (c *Consumer) sessionCursor(stored Cursor) Cursor { + switch { + case stored.Feed() == c.current && stored.Seq() > 0: + return stored + case c.ReplayFromStart: + return NewCursor(c.current, 0) + default: + return c.current.Live(c.now()) + } +} + func (c *Consumer) fill() { c.MaxHandlerAttempts = cmp.Or(c.MaxHandlerAttempts, defaultMaxHandlerAttempts) c.PoisonGrace = cmp.Or(c.PoisonGrace, defaultPoisonGrace) diff --git a/knotfeed/subscribe_test.go b/knotfeed/subscribe_test.go index bbdfe67e8..0080b43c4 100644 --- a/knotfeed/subscribe_test.go +++ b/knotfeed/subscribe_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "errors" + "fmt" "io" "log/slog" "net/http" @@ -62,12 +63,12 @@ func poisonConsumer(srv *httptest.Server, cursor *atomic.Int64, stored *atomic.I NoTLS: true, Logger: poisonTestLogger(), - LoadCursor: func(context.Context) (int64, error) { - return cursor.Load(), nil + LoadCursor: func(context.Context) (Cursor, error) { + return NewCursor(FeedAtproto, cursor.Load()), nil }, - StoreCursor: func(_ context.Context, seq int64) error { - cursor.Store(seq) - stored.Store(seq) + StoreCursor: func(_ context.Context, cur Cursor) error { + cursor.Store(cur.Seq()) + stored.Store(cur.Seq()) return nil }, Handle: func(context.Context, Message) error { @@ -114,3 +115,262 @@ func TestPoisonFrameHoldsWhileFailuresStayInsideTheGrace(t *testing.T) { t.Fatalf("cursor stores per attempt = %v, want all zero while inside the grace", seen) } } + +type pathLog struct{ paths chan string } + +func (p *pathLog) add(path string) { + select { + case p.paths <- path: + default: + } +} + +func (p *pathLog) take(t *testing.T, n int) []string { + t.Helper() + var seen []string + for range n { + select { + case path := <-p.paths: + seen = append(seen, path) + case <-time.After(10 * time.Second): + t.Fatalf("the consumer stopped after %v", seen) + } + } + return seen +} + +func wsServer(t *testing.T, refuse string, serve func(conn *websocket.Conn, connect int64)) (*httptest.Server, *pathLog) { + t.Helper() + upgrader := websocket.Upgrader{} + log := &pathLog{paths: make(chan string, 8)} + var connects atomic.Int64 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + log.add(r.URL.Path) + if r.URL.Path == refuse { + w.WriteHeader(http.StatusNotFound) + return + } + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + return + } + defer conn.Close() + serve(conn, connects.Add(1)) + _ = conn.WriteControl(websocket.CloseMessage, + websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""), time.Now().Add(time.Second)) + })) + t.Cleanup(srv.Close) + return srv, log +} + +func feedConsumer(t *testing.T, srv *httptest.Server, stored Cursor) *Consumer { + t.Helper() + c := &Consumer{ + Host: srv.Listener.Addr().String(), + NoTLS: true, + Logger: poisonTestLogger(), + LoadCursor: func(context.Context) (Cursor, error) { return stored, nil }, + StoreCursor: func(context.Context, Cursor) error { return nil }, + Handle: func(context.Context, Message) error { return nil }, + } + c.fill() + return c +} + +func runConsumer(t *testing.T, c *Consumer) { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + _ = c.Run(ctx) + }() + t.Cleanup(func() { + cancel() + <-done + }) +} + +func TestWrongKindFrameEndsSessionAsMismatch(t *testing.T) { + srv, _ := wsServer(t, "", func(conn *websocket.Conn, _ int64) { + _ = conn.WriteMessage(websocket.TextMessage, []byte(`{"nsid":"x","created":1}`)) + }) + c := feedConsumer(t, srv, Cursor{}) + + _, _, err := c.session(context.Background()) + + var mismatch feedMismatch + if !errors.As(err, &mismatch) { + t.Fatalf("session error = %v, want a feed mismatch", err) + } + if mismatch.feed != FeedAtproto { + t.Errorf("mismatch = %v, want the firehose we asked for", mismatch.feed) + } +} + +func TestMismatchTextSpellsOutBothFeeds(t *testing.T) { + got := feedMismatch{feed: FeedAtproto}.Error() + if got != "atproto firehose sent the knot event stream's frames" { + t.Errorf("mismatch reads %q", got) + } +} + +func TestConsumerLeavesFeedOnlyWhenKnotSaysTo(t *testing.T) { + for _, tt := range []struct { + name string + proven bool + err error + want Feed + }{ + {"mismatch on proven feed", true, feedMismatch{feed: FeedAtproto}, FeedLegacy}, + {"refused connection on unproven feed", false, connectError{errors.New("no route")}, FeedLegacy}, + {"refused connection on proven feed", true, connectError{errors.New("connection reset")}, FeedAtproto}, + {"session that failed for its own reasons", true, errors.New("boom"), FeedAtproto}, + } { + t.Run(tt.name, func(t *testing.T) { + c := &Consumer{Logger: poisonTestLogger()} + c.fill() + c.proven = tt.proven + + c.reactTo(tt.err) + + if c.current != tt.want { + t.Errorf("feed = %v, want %v", c.current, tt.want) + } + if switched := tt.want != FeedAtproto; switched && c.proven { + t.Error("new feed counts as proven before delivering anything") + } + }) + } +} + +func TestQuietKnotStillProvesFeedItAccepted(t *testing.T) { + srv, _ := wsServer(t, "", func(*websocket.Conn, int64) {}) + c := feedConsumer(t, srv, NewCursor(FeedLegacy, 11)) + + advanced, _, err := c.session(context.Background()) + + if advanced { + t.Error("empty session reported progress") + } + if err == nil { + t.Fatal("closed connection didn't error out") + } + if !c.proven { + t.Fatal("knot accepted the upgrade, and the feed still reads unproven") + } + + c.reactTo(connectError{errors.New("connection refused")}) + if c.current != FeedLegacy { + t.Errorf("feed = %v; we left a path the knot had already served after one refusal", c.current) + } +} + +func TestSwitchStaysUnprovenUntilNewFeedDelivers(t *testing.T) { + srv, paths := wsServer(t, eventsPath, func(conn *websocket.Conn, _ int64) { + _ = conn.WriteMessage(websocket.BinaryMessage, commitFrameForSeq(t, 500)) + _ = conn.WriteMessage(websocket.TextMessage, []byte(`{"nsid":"x","created":1}`)) + time.Sleep(time.Second) + }) + runConsumer(t, feedConsumer(t, srv, Cursor{})) + + tried := paths.take(t, 3) + + if tried[1] != eventsPath { + t.Fatalf("stray text frame sent us to %q, want the legacy path", tried[1]) + } + if tried[2] == eventsPath { + t.Error("we retried the refused legacy path, so the switch counted as proven too early") + } +} + +func TestStoredFeedIsAdoptedOnceAndNeverUndoesSwitch(t *testing.T) { + c := &Consumer{ + Host: "127.0.0.1:1", + NoTLS: true, + Logger: poisonTestLogger(), + LoadCursor: func(context.Context) (Cursor, error) { + return NewCursor(FeedAtproto, 500), nil + }, + StoreCursor: func(context.Context, Cursor) error { return nil }, + } + c.fill() + c.current = FeedLegacy + + _, _, err := c.session(context.Background()) + if c.current != FeedAtproto { + t.Fatalf("feed = %v, want the stored cursor's feed on the first session", c.current) + } + c.reactTo(err) + if c.current != FeedLegacy { + t.Fatalf("feed = %v, want a switch after the refused connection", c.current) + } + + for attempt := range 3 { + if _, _, _ = c.session(context.Background()); c.current != FeedLegacy { + t.Fatalf("session %d re-adopted the stored feed %v", attempt+2, c.current) + } + } +} + +func TestSubscriptionAsksForItsOwnFeedAndPosition(t *testing.T) { + frozen := time.Unix(0, 1788257313450261754) + const legacySeq = 1788245839553422000 + live := fmt.Sprint(frozen.UnixNano()) + for _, tt := range []struct { + name string + noTLS bool + feed Feed + replay bool + stored Cursor + want string + }{ + {"firehose position on the firehose", true, FeedAtproto, false, NewCursor(FeedAtproto, 42), + "ws://oyster.cafe/xrpc/com.atproto.sync.subscribeRepos?cursor=42"}, + {"nothing stored on the firehose", true, FeedAtproto, false, Cursor{}, + "ws://oyster.cafe/xrpc/com.atproto.sync.subscribeRepos"}, + {"legacy position on the legacy feed", false, FeedLegacy, false, NewCursor(FeedLegacy, legacySeq), + "wss://oyster.cafe/events?cursor=" + fmt.Sprint(legacySeq)}, + {"firehose position on the legacy feed", false, FeedLegacy, false, NewCursor(FeedAtproto, 42), + "wss://oyster.cafe/events?cursor=" + live}, + {"legacy position on the firehose", false, FeedAtproto, false, NewCursor(FeedLegacy, legacySeq), + "wss://oyster.cafe/xrpc/com.atproto.sync.subscribeRepos"}, + {"nothing stored on the legacy feed", false, FeedLegacy, false, Cursor{}, + "wss://oyster.cafe/events?cursor=" + live}, + {"nothing stored while replaying from the start", false, FeedLegacy, true, Cursor{}, + "wss://oyster.cafe/events?cursor=0"}, + } { + c := &Consumer{Host: "oyster.cafe", NoTLS: tt.noTLS, ReplayFromStart: tt.replay} + c.fill() + c.current = tt.feed + c.now = func() time.Time { return frozen } + if got := c.url(c.sessionCursor(tt.stored)); got != tt.want { + t.Errorf("%s asks for %q, want %q", tt.name, got, tt.want) + } + } +} + +func TestEveryFeedAndCursorRoundTripsItsText(t *testing.T) { + for _, feed := range []Feed{FeedAtproto, FeedLegacy} { + if parsed, err := ParseFeed(feed.Token()); err != nil || parsed != feed { + t.Errorf("ParseFeed(%q) = (%v, %v), want %v", feed.Token(), parsed, err, feed) + } + } + for _, want := range []Cursor{NewCursor(FeedAtproto, 500), NewCursor(FeedLegacy, 1788245839553422000), {}} { + got, err := ParseCursor(want.Encode()) + if err != nil || got != want { + t.Errorf("ParseCursor(%q) = (%v, %v), want %v", want.Encode(), got, err, want) + } + } + if got, err := ParseCursor("500"); err != nil || got != NewCursor(FeedAtproto, 500) { + t.Errorf("bare sequence parsed as (%v, %v), want the firehose at 500", got, err) + } + for _, raw := range []string{"carrier pigeon:7", "legacy:soon"} { + if _, err := ParseCursor(raw); err == nil { + t.Errorf("ParseCursor(%q) parsed anyway", raw) + } + } + if _, err := ParseFeed("carrier pigeon"); !errors.Is(err, ErrUnrecognizedFeed) { + t.Errorf("unrecognized token gave %v, want something callers can classify", err) + } +} diff --git a/knotmirror/db/db.go b/knotmirror/db/db.go index cfdd3d897..b9a7697fb 100644 --- a/knotmirror/db/db.go +++ b/knotmirror/db/db.go @@ -69,6 +69,7 @@ func Make(ctx context.Context, dbUrl string, maxConns int) (*sql.DB, error) { no_ssl boolean not null default false, status text not null default 'active', last_seq bigint not null default -1, + last_feed text not null default 'atproto', db_created_at timestamptz not null default now(), db_updated_at timestamptz not null default now(), diff --git a/knotmirror/db/hosts.go b/knotmirror/db/hosts.go index 2faec40e2..2a4c47c3f 100644 --- a/knotmirror/db/hosts.go +++ b/knotmirror/db/hosts.go @@ -7,48 +7,79 @@ import ( "fmt" "log" + "tangled.org/core/knotfeed" "tangled.org/core/knotmirror/models" ) func UpsertHost(ctx context.Context, e DBTX, host *models.Host) error { if _, err := e.ExecContext(ctx, - `insert into hosts (hostname, no_ssl, status, last_seq) - values ($1, $2, $3, $4) + `insert into hosts (hostname, no_ssl, status, last_seq, last_feed) + values ($1, $2, $3, $4, $5) on conflict(hostname) do update set - no_ssl = excluded.no_ssl, - status = excluded.status, - last_seq = excluded.last_seq + no_ssl = excluded.no_ssl, + status = excluded.status, + last_seq = excluded.last_seq, + last_feed = excluded.last_feed `, host.Hostname, host.NoSSL, host.Status, - host.LastSeq, + host.Cursor.Seq(), + host.Cursor.Feed().Token(), ); err != nil { return fmt.Errorf("upserting host: %w", err) } return nil } -func GetHost(ctx context.Context, e DBTX, hostname string) (*models.Host, error) { +type rowScanner interface { + Scan(dest ...any) error +} + +const hostColumns = `select hostname, no_ssl, status, last_seq, last_feed from hosts` + +func scanHost(row rowScanner) (models.Host, error) { var host models.Host - if err := e.QueryRowContext(ctx, - `select hostname, no_ssl, status, last_seq - from hosts where hostname = $1`, - hostname, - ).Scan( - &host.Hostname, - &host.NoSSL, - &host.Status, - &host.LastSeq, - ); err != nil { - if errors.Is(err, sql.ErrNoRows) { - return nil, nil - } + var seq int64 + var token string + if err := row.Scan(&host.Hostname, &host.NoSSL, &host.Status, &seq, &token); err != nil { + return host, err + } + host.Cursor = hostCursor(host.Hostname, seq, token) + return host, nil +} + +func GetHost(ctx context.Context, e DBTX, hostname string) (*models.Host, error) { + host, err := scanHost(e.QueryRowContext(ctx, hostColumns+` where hostname = $1`, hostname)) + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + if err != nil { return nil, err } return &host, nil } +func hostCursor(hostname string, seq int64, token string) knotfeed.Cursor { + feed, err := knotfeed.ParseFeed(token) + if err != nil { + log.Println("host cursor stored an unrecognized feed, resuming live", "host", hostname, "feed", token) + return knotfeed.Cursor{} + } + return knotfeed.NewCursor(feed, seq) +} + +func SetHostStatus(ctx context.Context, e DBTX, hostname string, status models.HostStatus) error { + if _, err := e.ExecContext(ctx, + `update hosts set status = $1 where hostname = $2`, + status, + hostname, + ); err != nil { + return fmt.Errorf("setting host status: %w", err) + } + return nil +} + func StoreCursors(ctx context.Context, e *sql.DB, cursors []models.HostCursor) error { tx, err := e.BeginTx(ctx, nil) if err != nil { @@ -56,27 +87,24 @@ func StoreCursors(ctx context.Context, e *sql.DB, cursors []models.HostCursor) e } defer tx.Rollback() for _, cur := range cursors { - if cur.LastSeq < 0 { + if cur.Cursor.Seq() < 0 { continue } if _, err := tx.ExecContext(ctx, - `update hosts set last_seq = $1 where hostname = $2`, - cur.LastSeq, + `update hosts set last_seq = $1, last_feed = $2 where hostname = $3`, + cur.Cursor.Seq(), + cur.Cursor.Feed().Token(), cur.Hostname, ); err != nil { - log.Println("failed to persist host cursor", "host", cur.Hostname, "lastSeq", cur.LastSeq, "err", err) + log.Println("couldn't persist host cursor", "host", cur.Hostname, + "feed", cur.Cursor.Feed(), "lastSeq", cur.Cursor.Seq(), "err", err) } } return tx.Commit() } func ListHosts(ctx context.Context, e DBTX, status models.HostStatus) ([]models.Host, error) { - rows, err := e.QueryContext(ctx, - `select hostname, no_ssl, status, last_seq - from hosts - where status = $1`, - status, - ) + rows, err := e.QueryContext(ctx, hostColumns+` where status = $1`, status) if err != nil { return nil, fmt.Errorf("querying hosts: %w", err) } @@ -84,13 +112,8 @@ func ListHosts(ctx context.Context, e DBTX, status models.HostStatus) ([]models. var hosts []models.Host for rows.Next() { - var host models.Host - if err := rows.Scan( - &host.Hostname, - &host.NoSSL, - &host.Status, - &host.LastSeq, - ); err != nil { + host, err := scanHost(rows) + if err != nil { return nil, fmt.Errorf("scanning row: %w", err) } hosts = append(hosts, host) diff --git a/knotmirror/db/hosts_test.go b/knotmirror/db/hosts_test.go new file mode 100644 index 000000000..8720d697a --- /dev/null +++ b/knotmirror/db/hosts_test.go @@ -0,0 +1,24 @@ +package db + +import ( + "testing" + + "tangled.org/core/knotfeed" +) + +func TestHostCursorKeepsFeedItWasStoredWith(t *testing.T) { + const legacySeq = 1788245839553422000 + for _, tt := range []struct { + token string + want knotfeed.Cursor + }{ + {knotfeed.FeedLegacy.Token(), knotfeed.NewCursor(knotfeed.FeedLegacy, legacySeq)}, + {knotfeed.FeedAtproto.Token(), knotfeed.NewCursor(knotfeed.FeedAtproto, legacySeq)}, + {"carrier pigeon", knotfeed.Cursor{}}, + } { + if got := hostCursor("oyster.cafe", legacySeq, tt.token); got != tt.want { + t.Errorf("hostCursor with feed %q = %v/%d, want %v/%d", + tt.token, got.Feed(), got.Seq(), tt.want.Feed(), tt.want.Seq()) + } + } +} diff --git a/knotmirror/db/migrations_list.go b/knotmirror/db/migrations_list.go index 256ec5246..5016f9c12 100644 --- a/knotmirror/db/migrations_list.go +++ b/knotmirror/db/migrations_list.go @@ -13,6 +13,16 @@ var Migrations = []Migration{ Name: "repos_pk_to_repo_did", Fn: reposPkToRepoDid, }, + { + Name: "add_last_feed_to_hosts", + Fn: addLastFeedToHosts, + }, +} + +func addLastFeedToHosts(ctx context.Context, tx *sql.Tx) error { + return execAll(ctx, tx, + `alter table hosts add column if not exists last_feed text not null default 'atproto'`, + ) } func reposPkToRepoDid(ctx context.Context, tx *sql.Tx) error { diff --git a/knotmirror/knotstream/knotstream.go b/knotmirror/knotstream/knotstream.go index f16f68ab9..624d1ecbd 100644 --- a/knotmirror/knotstream/knotstream.go +++ b/knotmirror/knotstream/knotstream.go @@ -53,7 +53,6 @@ func (s *KnotStream) SubscribeHost(ctx context.Context, hostname string, noSSL b Hostname: hostname, NoSSL: noSSL, Status: models.HostStatusActive, - LastSeq: 0, } if err := db.UpsertHost(ctx, s.db, host); err != nil { @@ -71,9 +70,8 @@ func (s *KnotStream) SubscribeHost(ctx context.Context, hostname string, noSSL b return fmt.Errorf("slurper: %w", err) } - host.Status = models.HostStatusActive - if err := db.UpsertHost(ctx, s.db, host); err != nil { - return fmt.Errorf("upserting host status to db: %w", err) + if err := db.SetHostStatus(ctx, s.db, host.Hostname, models.HostStatusActive); err != nil { + return fmt.Errorf("marking host active in db: %w", err) } return nil diff --git a/knotmirror/knotstream/scheduler.go b/knotmirror/knotstream/scheduler.go index 69721e3aa..129241961 100644 --- a/knotmirror/knotstream/scheduler.go +++ b/knotmirror/knotstream/scheduler.go @@ -6,6 +6,7 @@ import ( "sync" "time" + "github.com/bluesky-social/indigo/atproto/syntax" "tangled.org/core/knotfeed" "tangled.org/core/log" ) @@ -25,8 +26,8 @@ type ParallelScheduler struct { type Task struct { Key string - repoDid string - seq int64 + repoDid syntax.DID + cursor knotfeed.Cursor op knotfeed.RecordOp } diff --git a/knotmirror/knotstream/slurper.go b/knotmirror/knotstream/slurper.go index f6dca55fe..bd2b437fe 100644 --- a/knotmirror/knotstream/slurper.go +++ b/knotmirror/knotstream/slurper.go @@ -8,8 +8,8 @@ import ( "sync" "time" - "github.com/bluesky-social/indigo/atproto/syntax" "github.com/gorilla/websocket" + "github.com/samber/lo" "tangled.org/core/knotfeed" "tangled.org/core/knotmirror/config" "tangled.org/core/knotmirror/db" @@ -118,7 +118,7 @@ func (s *KnotSlurper) Subscribe(host models.Host) error { do := func(ctx context.Context, task *Task) error { err := s.ProcessEvent(ctx, task) if err == nil { - sub.MarkApplied(task.seq) + sub.MarkApplied(task.cursor) } return err } @@ -126,8 +126,7 @@ func (s *KnotSlurper) Subscribe(host models.Host) error { s.cfg.ConcurrencyPerHost, do, ) - sub.lastSeq.Store(host.LastSeq) - sub.appliedSeq.Store(host.LastSeq) + sub.Resume(host.Cursor) s.subs[host.Hostname] = sub ctx := context.TODO() @@ -162,7 +161,7 @@ func (s *KnotSlurper) runConsumer(ctx context.Context, host models.Host, sub *su connectedInbound.Inc() defer connectedInbound.Dec() - highWater := sub.LastSeq() + highWater := sub.Last() connectFailures := 0 consumer := &knotfeed.Consumer{ @@ -170,60 +169,52 @@ func (s *KnotSlurper) runConsumer(ctx context.Context, host models.Host, sub *su NoTLS: host.NoSSL, Dialer: s.dialer(host), Logger: l, - LoadCursor: func(context.Context) (int64, error) { - return sub.AppliedSeq(), nil + LoadCursor: func(context.Context) (knotfeed.Cursor, error) { + return sub.Applied(), nil }, - StoreCursor: func(_ context.Context, seq int64) error { - sub.lastSeq.Store(seq) + StoreCursor: func(_ context.Context, cursor knotfeed.Cursor) error { + sub.Seen(cursor) return nil }, Handle: func(ctx context.Context, msg knotfeed.Message) error { if msg.Type != knotfeed.TypeCommit || msg.Commit == nil { return nil } - ops := 0 - var first knotfeed.RecordOp - for _, op := range msg.Commit.Records { - if op.Collection != knotfeed.GitRefCollection { - continue - } - ops++ - if ops == 1 { - first = op - } - } - if ops == 0 { + refOps := lo.Filter(msg.Commit.Records, func(op knotfeed.RecordOp, _ int) bool { + return op.Collection == knotfeed.GitRefCollection + }) + if len(refOps) == 0 { return nil } - firehoseRefOpsReceived.Add(float64(ops)) + firehoseRefOpsReceived.Add(float64(len(refOps))) sub.scheduler.AddTask(ctx, &Task{ - Key: msg.Commit.Repo, + Key: msg.Commit.Repo.String(), repoDid: msg.Commit.Repo, - seq: msg.Commit.Seq, - op: first, + cursor: knotfeed.NewCursor(msg.Feed, msg.Commit.Seq), + op: refOps[0], }) return nil }, - OutdatedReplay: func(ctx context.Context) int64 { + OutdatedReplay: func(ctx context.Context, feed knotfeed.Feed) knotfeed.Cursor { if _, err := s.desynchronizeHostRepos(ctx, host); err != nil { l.Warn("couldn't mark repos desynchronized, retrying on the next outdated notice") - return sub.LastSeq() + return sub.Last() } - return 0 + return feed.Live(time.Now()) }, OnConnectError: func(connectErr error) { - if seq := sub.LastSeq(); seq > highWater { - highWater = seq + if seen := sub.Last(); seen.Feed() != highWater.Feed() || seen.Seq() > highWater.Seq() { + highWater = seen connectFailures = 0 } connectFailures++ - l.Warn("dialing failed", "err", connectErr, "failures", connectFailures) + l.Warn("couldn't reach knot", "err", connectErr, "failures", connectFailures) if connectFailures <= maxConnectFailures { return } l.Warn("host doesn't appear to be online, disabling for now") host.Status = models.HostStatusOffline - host.LastSeq = sub.AppliedSeq() + host.Cursor = sub.Applied() if err := db.UpsertHost(ctx, s.db, &host); err != nil { l.Error("failed to update host status", "err", err) } @@ -253,14 +244,8 @@ func (s *KnotSlurper) ProcessEvent(ctx context.Context, task *Task) error { firehoseRefOpsSkipped.Inc() return nil } - repoDid, err := syntax.ParseDID(task.repoDid) - if err != nil { - l.Warn("skipping ref op: commit frame has invalid repo did", "err", err) - firehoseRefOpsSkipped.Inc() - return nil - } - curr, err := db.GetRepoByRepoDid(ctx, s.db, repoDid) + curr, err := db.GetRepoByRepoDid(ctx, s.db, task.repoDid) if err != nil { return fmt.Errorf("failed to get repo '%s': %w", task.repoDid, err) } diff --git a/knotmirror/knotstream/subscription.go b/knotmirror/knotstream/subscription.go index 1231aef68..8a3227467 100644 --- a/knotmirror/knotstream/subscription.go +++ b/knotmirror/knotstream/subscription.go @@ -1,40 +1,60 @@ package knotstream import ( - "sync/atomic" + "sync" + "tangled.org/core/knotfeed" "tangled.org/core/knotmirror/models" ) type subscription struct { hostname string - lastSeq atomic.Int64 - appliedSeq atomic.Int64 + mu sync.Mutex + last knotfeed.Cursor + applied knotfeed.Cursor scheduler *ParallelScheduler } -func (s *subscription) LastSeq() int64 { - return s.lastSeq.Load() +func (s *subscription) Last() knotfeed.Cursor { + s.mu.Lock() + defer s.mu.Unlock() + return s.last } -func (s *subscription) AppliedSeq() int64 { - return s.appliedSeq.Load() +func (s *subscription) Applied() knotfeed.Cursor { + s.mu.Lock() + defer s.mu.Unlock() + return s.applied } -func (s *subscription) MarkApplied(seq int64) { - for { - current := s.appliedSeq.Load() - if seq <= current || s.appliedSeq.CompareAndSwap(current, seq) { - return - } +func (s *subscription) Resume(cursor knotfeed.Cursor) { + s.mu.Lock() + defer s.mu.Unlock() + s.last, s.applied = cursor, cursor +} + +func (s *subscription) Seen(cursor knotfeed.Cursor) { + s.mu.Lock() + defer s.mu.Unlock() + if cursor.Feed() != s.last.Feed() { + s.applied = knotfeed.NewCursor(cursor.Feed(), 0) + } + s.last = cursor +} + +func (s *subscription) MarkApplied(cursor knotfeed.Cursor) { + s.mu.Lock() + defer s.mu.Unlock() + if cursor.Feed() == s.applied.Feed() { + s.applied = knotfeed.NewCursor(s.applied.Feed(), max(s.applied.Seq(), cursor.Seq())) } } func (s *subscription) HostCursor() models.HostCursor { return models.HostCursor{ Hostname: s.hostname, - LastSeq: s.AppliedSeq(), + Cursor: s.Applied(), } } diff --git a/knotmirror/knotstream/subscription_test.go b/knotmirror/knotstream/subscription_test.go new file mode 100644 index 000000000..520ab106f --- /dev/null +++ b/knotmirror/knotstream/subscription_test.go @@ -0,0 +1,65 @@ +package knotstream + +import ( + "testing" + + "github.com/samber/lo" + "tangled.org/core/knotfeed" +) + +func TestSubscriptionTracksOnePositionPerFeed(t *testing.T) { + const legacySeq = 1788245839553422000 + atproto := func(seq int64) knotfeed.Cursor { + return knotfeed.NewCursor(knotfeed.FeedAtproto, seq) + } + legacy := func(seq int64) knotfeed.Cursor { + return knotfeed.NewCursor(knotfeed.FeedLegacy, seq) + } + + for _, tt := range []struct { + name string + resume knotfeed.Cursor + seen []knotfeed.Cursor + applied []knotfeed.Cursor + wantLast knotfeed.Cursor + wantApplied knotfeed.Cursor + }{ + { + name: "applied rises to the high water mark", resume: atproto(10), + applied: []knotfeed.Cursor{atproto(40), atproto(20)}, + wantLast: atproto(10), wantApplied: atproto(40), + }, + { + name: "straggler from the feed we left is dropped", resume: legacy(legacySeq), + seen: []knotfeed.Cursor{atproto(500)}, + applied: []knotfeed.Cursor{legacy(legacySeq + 1)}, + wantLast: atproto(500), wantApplied: atproto(0), + }, + { + name: "changing feeds clears the applied position", resume: legacy(legacySeq), + seen: []knotfeed.Cursor{atproto(7)}, + applied: []knotfeed.Cursor{atproto(7)}, + wantLast: atproto(7), wantApplied: atproto(7), + }, + { + name: "cursor from another feed resets the subscription", resume: atproto(500), + seen: []knotfeed.Cursor{legacy(legacySeq)}, + wantLast: legacy(legacySeq), wantApplied: legacy(0), + }, + } { + t.Run(tt.name, func(t *testing.T) { + sub := subscription{hostname: "oyster.cafe"} + sub.Resume(tt.resume) + lo.ForEach(tt.seen, func(c knotfeed.Cursor, _ int) { sub.Seen(c) }) + lo.ForEach(tt.applied, func(c knotfeed.Cursor, _ int) { sub.MarkApplied(c) }) + + if got := sub.Last(); got != tt.wantLast { + t.Errorf("last = %v/%d, want %v/%d", got.Feed(), got.Seq(), tt.wantLast.Feed(), tt.wantLast.Seq()) + } + if got := sub.HostCursor(); got.Hostname != "oyster.cafe" || got.Cursor != tt.wantApplied { + t.Errorf("host cursor = %s at %v/%d, want oyster.cafe at %v/%d", + got.Hostname, got.Cursor.Feed(), got.Cursor.Seq(), tt.wantApplied.Feed(), tt.wantApplied.Seq()) + } + }) + } +} diff --git a/knotmirror/models/models.go b/knotmirror/models/models.go index da21d292b..911bf153a 100644 --- a/knotmirror/models/models.go +++ b/knotmirror/models/models.go @@ -5,6 +5,7 @@ import ( "github.com/bluesky-social/indigo/atproto/syntax" "tangled.org/core/api/tangled" + "tangled.org/core/knotfeed" ) type Repo struct { @@ -65,14 +66,14 @@ const ( type HostCursor struct { Hostname string - LastSeq int64 + Cursor knotfeed.Cursor } type Host struct { Hostname string NoSSL bool Status HostStatus - LastSeq int64 + Cursor knotfeed.Cursor } type HostStatus string diff --git a/knotmirror/templates/hosts.html b/knotmirror/templates/hosts.html index d7d72701a..8c6ca8df9 100644 --- a/knotmirror/templates/hosts.html +++ b/knotmirror/templates/hosts.html @@ -25,6 +25,7 @@ Hostname SSL Status + Feed Last Seq @@ -34,10 +35,11 @@ {{.Hostname}} {{if .NoSSL}}False{{else}}True{{end}} {{.Status}} - {{.LastSeq}} + {{.Cursor.Feed}} + {{.Cursor.Seq}} {{else}} - No hosts registered. + No hosts registered. {{end}} diff --git a/spindle/db/db.go b/spindle/db/db.go index c5c1c7a90..002d5e6b7 100644 --- a/spindle/db/db.go +++ b/spindle/db/db.go @@ -67,7 +67,8 @@ func Make(ctx context.Context, dbPath string) (*DB, error) { create table if not exists feed_cursors ( knot text primary key, - seq integer not null + seq integer not null, + feed text not null default 'atproto' ); create table if not exists feed_refs ( @@ -968,6 +969,24 @@ func runMigrations(_ context.Context, conn *sql.Conn, logger *slog.Logger) error return err } + if err := orm.RunMigration(conn, logger, "add-feed-to-feed-cursors", func(tx *sql.Tx) error { + var present int + if err := tx.QueryRow( + `select count(*) from pragma_table_info('feed_cursors') where name = 'feed'`, + ).Scan(&present); err != nil { + return err + } + if present != 0 { + return nil + } + _, err := tx.Exec( + `alter table feed_cursors add column feed text not null default 'atproto'`, + ) + return err + }); err != nil { + return err + } + return nil } diff --git a/spindle/db/feed.go b/spindle/db/feed.go index 9300d9774..d79fb2a60 100644 --- a/spindle/db/feed.go +++ b/spindle/db/feed.go @@ -3,65 +3,81 @@ package db import ( "database/sql" "errors" + "fmt" "github.com/bluesky-social/indigo/atproto/syntax" + "tangled.org/core/knotfeed" ) -func (d *DB) LoadFeedCursor(knot string) (int64, error) { +func (d *DB) LoadFeedCursor(knot string) (knotfeed.Cursor, error) { var seq int64 - err := d.QueryRow(`select seq from feed_cursors where knot = ?`, knot).Scan(&seq) + var token string + err := d.QueryRow( + `select seq, feed from feed_cursors where knot = ?`, knot, + ).Scan(&seq, &token) if errors.Is(err, sql.ErrNoRows) { - return 0, nil + return knotfeed.Cursor{}, nil + } + if err != nil { + return knotfeed.Cursor{}, err } - return seq, err + feed, err := knotfeed.ParseFeed(token) + if err != nil { + return knotfeed.Cursor{}, fmt.Errorf("feed cursor for %s: %w", knot, err) + } + return knotfeed.NewCursor(feed, seq), nil } -func (d *DB) StoreFeedCursor(knot string, seq int64) error { +func (d *DB) StoreFeedCursor(knot string, cursor knotfeed.Cursor) error { _, err := d.Exec( - `insert into feed_cursors (knot, seq) values (?, ?) - on conflict(knot) do update set seq = excluded.seq`, - knot, seq, + `insert into feed_cursors (knot, seq, feed) values (?, ?, ?) + on conflict(knot) do update set seq = excluded.seq, feed = excluded.feed`, + knot, cursor.Seq(), cursor.Feed().Token(), ) return err } -func (d *DB) FeedRefSha(repoDid syntax.DID, rkey string) (string, bool, error) { - var sha string +func (d *DB) FeedRefSha(repoDid syntax.DID, rkey syntax.RecordKey) (knotfeed.ObjectID, bool, error) { + var raw string err := d.QueryRow( `select sha from feed_refs where repo_did = ? and rkey = ?`, - repoDid.String(), rkey, - ).Scan(&sha) + repoDid.String(), rkey.String(), + ).Scan(&raw) if errors.Is(err, sql.ErrNoRows) { - return "", false, nil + return knotfeed.ObjectID{}, false, nil } if err != nil { - return "", false, err + return knotfeed.ObjectID{}, false, err + } + sha, ok := knotfeed.ParseObjectID(raw) + if !ok { + return knotfeed.ObjectID{}, false, fmt.Errorf("ref state for %s at %s stored sha %q", repoDid, rkey, raw) } return sha, true, nil } -func (d *DB) PutFeedRef(repoDid syntax.DID, rkey, sha string) error { +func (d *DB) PutFeedRef(repoDid syntax.DID, rkey syntax.RecordKey, sha knotfeed.ObjectID) error { _, err := d.Exec( `insert into feed_refs (repo_did, rkey, sha) values (?, ?, ?) on conflict(repo_did, rkey) do update set sha = excluded.sha`, - repoDid.String(), rkey, sha, + repoDid.String(), rkey.String(), sha.String(), ) return err } -func (d *DB) SeedFeedRef(repoDid syntax.DID, rkey, sha string) error { +func (d *DB) SeedFeedRef(repoDid syntax.DID, rkey syntax.RecordKey, sha knotfeed.ObjectID) error { _, err := d.Exec( `insert into feed_refs (repo_did, rkey, sha) values (?, ?, ?) on conflict(repo_did, rkey) do nothing`, - repoDid.String(), rkey, sha, + repoDid.String(), rkey.String(), sha.String(), ) return err } -func (d *DB) DeleteFeedRef(repoDid syntax.DID, rkey string) error { +func (d *DB) DeleteFeedRef(repoDid syntax.DID, rkey syntax.RecordKey) error { _, err := d.Exec( `delete from feed_refs where repo_did = ? and rkey = ?`, - repoDid.String(), rkey, + repoDid.String(), rkey.String(), ) return err } diff --git a/spindle/feed/feed.go b/spindle/feed/feed.go index dc2f167a2..a3105ae3b 100644 --- a/spindle/feed/feed.go +++ b/spindle/feed/feed.go @@ -4,16 +4,17 @@ import ( "context" "log/slog" "sync" + "time" "tangled.org/core/hostutil" knotfeed "tangled.org/core/knotfeed" ) type Hooks struct { - LoadCursor func(ctx context.Context, knot string) (int64, error) - StoreCursor func(ctx context.Context, knot string, seq int64) error + LoadCursor func(ctx context.Context, knot string) (knotfeed.Cursor, error) + StoreCursor func(ctx context.Context, knot string, cursor knotfeed.Cursor) error Handle func(ctx context.Context, knot string, msg knotfeed.Message) error - OutdatedReplay func(ctx context.Context, knot string) int64 + OutdatedReplay func(ctx context.Context, knot string, feed knotfeed.Feed) knotfeed.Cursor OnConnectError func(knot string, err error) } @@ -64,20 +65,20 @@ func (f *Feed) Subscribe(ctx context.Context, knot string) { Host: host, NoTLS: noTLS, Logger: f.logger, - LoadCursor: func(ctx context.Context) (int64, error) { + LoadCursor: func(ctx context.Context) (knotfeed.Cursor, error) { return f.hooks.LoadCursor(ctx, knot) }, - StoreCursor: func(ctx context.Context, seq int64) error { - return f.hooks.StoreCursor(ctx, knot, seq) + StoreCursor: func(ctx context.Context, cursor knotfeed.Cursor) error { + return f.hooks.StoreCursor(ctx, knot, cursor) }, Handle: func(ctx context.Context, msg knotfeed.Message) error { return f.hooks.Handle(ctx, knot, msg) }, - OutdatedReplay: func(ctx context.Context) int64 { + OutdatedReplay: func(ctx context.Context, feed knotfeed.Feed) knotfeed.Cursor { if f.hooks.OutdatedReplay == nil { - return 0 + return feed.Live(time.Now()) } - return f.hooks.OutdatedReplay(ctx, knot) + return f.hooks.OutdatedReplay(ctx, knot, feed) }, } if f.hooks.OnConnectError != nil { diff --git a/spindle/feed/feed_test.go b/spindle/feed/feed_test.go index d3ddb90c5..7781fad9d 100644 --- a/spindle/feed/feed_test.go +++ b/spindle/feed/feed_test.go @@ -7,23 +7,25 @@ import ( "log/slog" "testing" "time" + + "tangled.org/core/knotfeed" ) func TestFeedRefcountsAndReleasesSubscriptions(t *testing.T) { entered := make(chan string, 8) held := make(chan struct{}) f := New(slog.New(slog.NewTextHandler(io.Discard, nil)), Hooks{ - LoadCursor: func(ctx context.Context, knot string) (int64, error) { + LoadCursor: func(ctx context.Context, knot string) (knotfeed.Cursor, error) { select { case entered <- knot: case <-ctx.Done(): - return 0, ctx.Err() + return knotfeed.Cursor{}, ctx.Err() } select { case <-held: case <-ctx.Done(): } - return 0, errors.New("no network in tests") + return knotfeed.Cursor{}, errors.New("no network in tests") }, }) ctx, cancel := context.WithCancel(context.Background()) @@ -59,9 +61,9 @@ func TestFeedRefcountsAndReleasesSubscriptions(t *testing.T) { func TestFeedRejectsUnsubscribableHost(t *testing.T) { entered := make(chan string, 1) f := New(slog.New(slog.NewTextHandler(io.Discard, nil)), Hooks{ - LoadCursor: func(ctx context.Context, knot string) (int64, error) { + LoadCursor: func(ctx context.Context, knot string) (knotfeed.Cursor, error) { entered <- knot - return 0, errors.New("no network in tests") + return knotfeed.Cursor{}, errors.New("no network in tests") }, }) ctx, cancel := context.WithCancel(context.Background()) diff --git a/spindle/knotfeed.go b/spindle/knotfeed.go index 1f1641160..6a3eb9cb5 100644 --- a/spindle/knotfeed.go +++ b/spindle/knotfeed.go @@ -26,7 +26,7 @@ import ( "tangled.org/core/workflow" ) -const collaboratorInviteNSID = "sh.tangled.repo.collaboratorInvite" +const collaboratorInviteNSID syntax.NSID = tangled.RepoCollaboratorInviteNSID const ( reconcileInterval = 10 * time.Minute @@ -46,8 +46,8 @@ const ( var knotHTTPClient = &http.Client{Timeout: 30 * time.Second} type refRecord struct { - Rkey string - Sha string + Rkey syntax.RecordKey + Sha knotfeed.ObjectID } func (s *Spindle) handleKnotFeed(ctx context.Context, knot string, msg knotfeed.Message) error { @@ -58,7 +58,7 @@ func (s *Spindle) handleKnotFeed(ctx context.Context, knot string, msg knotfeed. var invites []syntax.DID seen := make(map[syntax.DID]struct{}) for _, op := range msg.Commit.Records { - repoDid := syntax.DID(msg.Commit.Repo) + repoDid := msg.Commit.Repo switch op.Collection { case knotfeed.GitRefCollection: if err := s.handleRefOp(ctx, knot, repoDid, op); err != nil { @@ -112,10 +112,6 @@ func (s *Spindle) handleRefOp(ctx context.Context, knot string, repoDid syntax.D return nil } - if err := s.ensureRefState(ctx, knot, repoDid); err != nil { - return fmt.Errorf("bootstrapping ref state: %w", err) - } - if op.Deleted() { if err := s.db.DeleteFeedRef(repoDid, op.Rkey); err != nil { return fmt.Errorf("forgetting ref state: %w", err) @@ -129,9 +125,9 @@ func (s *Spindle) handleRefOp(ctx context.Context, knot string, repoDid syntax.D return fmt.Errorf("decoding ref record: %w", err) } - oldSha, _, err := s.db.FeedRefSha(repoDid, op.Rkey) + oldSha, err := s.priorSha(ctx, knot, repoDid, op) if err != nil { - return fmt.Errorf("reading ref state: %w", err) + return err } if kgit.HasSkipCIPushOption(record.PushOptions) { if err := s.db.PutFeedRef(repoDid, op.Rkey, record.Sha); err != nil { @@ -143,7 +139,7 @@ func (s *Spindle) handleRefOp(ctx context.Context, knot string, repoDid syntax.D repoCloneUri := s.newRepoCloneUrl(knot, repoDid) repoPath := s.newRepoPath(repoDid) - if err := gitutil.SparseSync(ctx, repoCloneUri, repoPath, record.Sha, sparseWorkflowDir); err != nil { + if err := gitutil.SparseSync(ctx, repoCloneUri, repoPath, record.Sha.String(), sparseWorkflowDir); err != nil { return fmt.Errorf("sync git repo: %w", err) } l.Info("synced git repo") @@ -163,13 +159,13 @@ func (s *Spindle) handleRefOp(ctx context.Context, knot string, repoDid syntax.D Kind: string(workflow.TriggerKindPush), Push: &tangled.Pipeline_PushTriggerData{ Ref: refname, - OldSha: oldSha, - NewSha: record.Sha, + OldSha: oldSha.String(), + NewSha: record.Sha.String(), }, Repo: triggerRepo, } - pipelineId, err := s.runPipeline(ctx, repoDid, trigger, changedFiles, repoCloneUri, repoPath, record.Sha, nil, triggerRepo) + pipelineId, err := s.runPipeline(ctx, repoDid, trigger, changedFiles, repoCloneUri, repoPath, record.Sha.String(), nil, triggerRepo) if err != nil { return err } @@ -185,6 +181,20 @@ func isMaterializedRef(refname string) bool { return strings.HasPrefix(refname, "refs/heads/") || strings.HasPrefix(refname, "refs/tags/") } +func (s *Spindle) priorSha(ctx context.Context, knot string, repoDid syntax.DID, op knotfeed.RecordOp) (knotfeed.ObjectID, error) { + if sha, ok := op.Prior.Sha(); ok { + return sha, nil + } + if err := s.ensureRefState(ctx, knot, repoDid); err != nil { + return knotfeed.ObjectID{}, fmt.Errorf("bootstrapping ref state: %w", err) + } + sha, _, err := s.db.FeedRefSha(repoDid, op.Rkey) + if err != nil { + return knotfeed.ObjectID{}, fmt.Errorf("reading ref state: %w", err) + } + return sha, nil +} + func (s *Spindle) ensureRefState(ctx context.Context, knot string, repoDid syntax.DID) error { if _, seeded := s.refStateSeeded.LoadOrStore(repoDid.String(), struct{}{}); seeded { return nil @@ -203,16 +213,16 @@ func (s *Spindle) ensureRefState(ctx context.Context, knot string, repoDid synta return nil } -func changedFilesUnderBudget(l *slog.Logger, repoPath, oldSha, newSha string) []string { - if oldSha == "" { +func changedFilesUnderBudget(l *slog.Logger, repoPath string, oldSha, newSha knotfeed.ObjectID) []string { + if oldSha.IsZero() { return nil } - gr, err := kgit.Open(repoPath, newSha) + gr, err := kgit.Open(repoPath, newSha.String()) if err != nil { l.Warn("cannot open synced repo for changed files", "err", err) return nil } - paths, err := gr.ChangedFilesBetween(oldSha, newSha) + paths, err := gr.ChangedFilesBetween(oldSha.String(), newSha.String()) if err != nil { l.Warn("changed files unavailable between revisions", "oldSha", oldSha, "newSha", newSha, "err", err) return nil @@ -319,11 +329,12 @@ func (s *Spindle) reconcileAllCollaborators(ctx context.Context) { } } -func (s *Spindle) feedOutdatedReplay(ctx context.Context, knot string) int64 { +func (s *Spindle) feedOutdatedReplay(ctx context.Context, knot string, feed knotfeed.Feed) knotfeed.Cursor { + live := feed.Live(time.Now()) repos, err := s.db.AllRepos() if err != nil { s.l.Warn("failed to load repos after outdated cursor", "knot", knot, "err", err) - return 0 + return live } reset := 0 for _, repo := range repos { @@ -340,7 +351,7 @@ func (s *Spindle) feedOutdatedReplay(ctx context.Context, knot string) int64 { if reset > 0 { s.l.Warn("knot cannot replay from our cursor, ref state will be re-seeded live", "knot", knot, "repos", reset) } - return 0 + return live } func (s *Spindle) wipeFeedRefs(ctx context.Context, repoDid syntax.DID) error { @@ -410,43 +421,39 @@ func (s *Spindle) refRecords(ctx context.Context, knot string, repoDid syntax.DI return s.listRefRecords(ctx, knot, repoDid) } - base, err := knotEndpoint(knot, "/xrpc/com.atproto.repo.listRecords") + base, err := knotEndpoint(knot, "/xrpc/sh.tangled.git.listRefs") if err != nil { return nil, err } var refs []refRecord cursor := "" - pages := 0 - for { - pages++ - if pages > maxListPages { - return nil, fmt.Errorf("ref records for %s exceed %d pages", repoDid, maxListPages) + for page := 0; ; page++ { + if page >= maxListPages { + return nil, fmt.Errorf("refs for %s exceed %d pages", repoDid, maxListPages) } q := url.Values{} q.Set("repo", repoDid.String()) - q.Set("collection", knotfeed.GitRefCollection) - q.Set("limit", "100") + q.Set("limit", "1000") if cursor != "" { q.Set("cursor", cursor) } var out struct { - Records []struct { - Uri string `json:"uri"` - Value struct { - Sha string `json:"sha"` - } `json:"value"` - } `json:"records"` + Refs []struct { + Ref string `json:"ref"` + Sha string `json:"sha"` + } `json:"refs"` Cursor string `json:"cursor"` } if err := getJSON(ctx, base+"?"+q.Encode(), &out); err != nil { return nil, err } - for _, rec := range out.Records { - uri, err := syntax.ParseATURI(rec.Uri) - if err != nil { - return nil, fmt.Errorf("parsing ref record uri %s: %w", rec.Uri, err) + for _, ref := range out.Refs { + rkey, named := knotfeed.EscapeRefname(ref.Ref) + sha, isObject := knotfeed.ParseObjectID(ref.Sha) + if !named || !isObject { + continue } - refs = append(refs, refRecord{Rkey: uri.RecordKey().String(), Sha: rec.Value.Sha}) + refs = append(refs, refRecord{Rkey: rkey, Sha: sha}) } if out.Cursor == "" { return refs, nil diff --git a/spindle/knotfeed_test.go b/spindle/knotfeed_test.go index 9da85fa82..24ca71c0a 100644 --- a/spindle/knotfeed_test.go +++ b/spindle/knotfeed_test.go @@ -21,6 +21,7 @@ import ( "tangled.org/core/knotfeed" "tangled.org/core/rbac" "tangled.org/core/spindle/db" + "tangled.org/core/tapc" ) const ( @@ -28,11 +29,24 @@ const ( testForeign = "barnacle.nel.pet" testRepoDid = syntax.DID("did:plc:limpet") testSubject = syntax.DID("did:plc:boltless") - testRkeyMain = "refs~2fheads~2fmain" - testShaOld = "1111111111111111111111111111111111111111" - testShaNew = "2222222222222222222222222222222222222222" + testRkeyMain = syntax.RecordKey("refs~2fheads~2fmain") + testHexOld = "1111111111111111111111111111111111111111" + testHexNew = "2222222222222222222222222222222222222222" ) +var ( + testShaOld = mustObjectID(testHexOld) + testShaNew = mustObjectID(testHexNew) +) + +func mustObjectID(hex string) knotfeed.ObjectID { + sha, ok := knotfeed.ParseObjectID(hex) + if !ok { + panic("test fixture " + hex + " is not an object id") + } + return sha +} + func newTestFeedSpindle(t *testing.T) *Spindle { t.Helper() d, e := newTestSpindleDB(t) @@ -44,7 +58,7 @@ func newTestFeedSpindle(t *testing.T) *Spindle { return s } -func feedCommit(repo string, ops ...knotfeed.RecordOp) knotfeed.Message { +func feedCommit(repo syntax.DID, ops ...knotfeed.RecordOp) knotfeed.Message { return knotfeed.Message{ Type: knotfeed.TypeCommit, Commit: &knotfeed.Commit{ @@ -147,8 +161,8 @@ func TestKnotFeedRefOps(t *testing.T) { t.Fatalf("seed ref state: %v", err) } - if err := s.handleKnotFeed(context.Background(), testKnot, feedCommit(testRepoDid.String(), - knotfeed.RecordOp{Action: "delete", Collection: knotfeed.GitRefCollection, Rkey: testRkeyMain}, + if err := s.handleKnotFeed(context.Background(), testKnot, feedCommit(testRepoDid, + knotfeed.RecordOp{Action: tapc.RecordDeleteAction, Collection: knotfeed.GitRefCollection, Rkey: testRkeyMain}, )); err != nil { t.Fatalf("handleKnotFeed: %v", err) } @@ -166,8 +180,8 @@ func TestKnotFeedRefOps(t *testing.T) { t.Fatalf("seed ref state: %v", err) } - if err := s.handleKnotFeed(context.Background(), testKnot, feedCommit(testRepoDid.String(), - knotfeed.RecordOp{Action: "create", Collection: knotfeed.GitRefCollection, Rkey: testRkeyMain, Bytes: encodeRefRecord(testShaNew, "skip-ci")}, + if err := s.handleKnotFeed(context.Background(), testKnot, feedCommit(testRepoDid, + knotfeed.RecordOp{Action: tapc.RecordCreateAction, Collection: knotfeed.GitRefCollection, Rkey: testRkeyMain, Bytes: encodeRefRecord(testHexNew, "skip-ci")}, )); err != nil { t.Fatalf("handleKnotFeed: %v", err) } @@ -179,11 +193,11 @@ func TestKnotFeedRefOps(t *testing.T) { t.Run("a create op that never materializes skips bootstrap", func(t *testing.T) { for _, tc := range []struct { name string - repo string - rkey string + repo syntax.DID + rkey syntax.RecordKey }{ {"unknown repo", "did:plc:unknown", testRkeyMain}, - {"ref outside heads and tags", testRepoDid.String(), "refs~2fnotes~2fwip"}, + {"ref outside heads and tags", testRepoDid, "refs~2fnotes~2fwip"}, } { t.Run(tc.name, func(t *testing.T) { s := newTestFeedSpindle(t) @@ -193,7 +207,7 @@ func TestKnotFeedRefOps(t *testing.T) { } if err := s.handleKnotFeed(context.Background(), testKnot, feedCommit(tc.repo, - knotfeed.RecordOp{Action: "create", Collection: knotfeed.GitRefCollection, Rkey: tc.rkey, Bytes: encodeRefRecord(testShaNew)}, + knotfeed.RecordOp{Action: tapc.RecordCreateAction, Collection: knotfeed.GitRefCollection, Rkey: tc.rkey, Bytes: encodeRefRecord(testHexNew)}, )); err != nil { t.Fatalf("handleKnotFeed: %v", err) } @@ -207,8 +221,8 @@ func TestKnotFeedRefOps(t *testing.T) { return nil, nil } - if err := s.handleKnotFeed(context.Background(), testKnot, feedCommit(testRepoDid.String(), - knotfeed.RecordOp{Action: "create", Collection: knotfeed.GitRefCollection, Rkey: testRkeyMain, Bytes: []byte{0xff}}, + if err := s.handleKnotFeed(context.Background(), testKnot, feedCommit(testRepoDid, + knotfeed.RecordOp{Action: tapc.RecordCreateAction, Collection: knotfeed.GitRefCollection, Rkey: testRkeyMain, Bytes: []byte{0xff}}, )); err == nil { t.Fatal("expected an error for an undecodable ref record") } @@ -222,8 +236,8 @@ func TestKnotFeedRefOps(t *testing.T) { return nil, nil } - if err := s.handleKnotFeed(context.Background(), testKnot, feedCommit(testRepoDid.String(), - knotfeed.RecordOp{Action: "create", Collection: collaboratorInviteNSID, Rkey: testSubject.String()}, + if err := s.handleKnotFeed(context.Background(), testKnot, feedCommit(testRepoDid, + knotfeed.RecordOp{Action: tapc.RecordCreateAction, Collection: collaboratorInviteNSID, Rkey: syntax.RecordKey(testSubject)}, )); err != nil { t.Fatalf("handleKnotFeed: %v", err) } @@ -235,7 +249,7 @@ func TestKnotFeedRefOps(t *testing.T) { func TestKnotFeedSeedsRefStateOnce(t *testing.T) { s := newTestFeedSpindle(t) - rkeyTag := "refs~2ftags~2fv1" + rkeyTag := syntax.RecordKey("refs~2ftags~2fv1") fetched := 0 s.listRefRecords = func(ctx context.Context, knot string, repoDid syntax.DID) ([]refRecord, error) { if fetched++; fetched == 1 { @@ -250,7 +264,10 @@ func TestKnotFeedSeedsRefStateOnce(t *testing.T) { if err := s.ensureRefState(context.Background(), testKnot, testRepoDid); err != nil { t.Fatalf("ensureRefState: %v", err) } - for _, want := range []struct{ rkey, sha string }{ + for _, want := range []struct { + rkey syntax.RecordKey + sha knotfeed.ObjectID + }{ {testRkeyMain, testShaOld}, {rkeyTag, testShaNew}, } { @@ -293,7 +310,7 @@ func TestAdmitChangedFilesUnderBudget(t *testing.T) { func TestChangedFilesUnderBudgetGuardsFirstSight(t *testing.T) { quiet := slog.New(slog.NewTextHandler(io.Discard, nil)) - if got := changedFilesUnderBudget(quiet, "/nonexistent/repo", "", "2222222222222222222222222222222222222222"); got != nil { + if got := changedFilesUnderBudget(quiet, "/nonexistent/repo", knotfeed.ObjectID{}, testShaNew); got != nil { t.Errorf("first sight of a ref = %v, want nil without touching the repo", got) } } @@ -343,10 +360,10 @@ func TestChangedFilesUnderBudgetWithRealRepo(t *testing.T) { t.Fatalf("write %s: %v", name, err) } } - commit := func() string { + commit := func() knotfeed.ObjectID { run("-C", repoPath, "add", "-A") run("-C", repoPath, "commit", "-q", "-m", "wip") - return run("-C", repoPath, "rev-parse", "HEAD") + return mustObjectID(run("-C", repoPath, "rev-parse", "HEAD")) } write("README.md", "one\n") @@ -360,8 +377,8 @@ func TestChangedFilesUnderBudgetWithRealRepo(t *testing.T) { quiet := slog.New(slog.NewTextHandler(io.Discard, nil)) for _, tc := range []struct { name string - oldSha string - newSha string + oldSha knotfeed.ObjectID + newSha knotfeed.ObjectID want []string }{ {"a push commit lists its added and modified files", shaFirst, shaSecond, []string{".tangled/workflows/ci.yaml", "src/app.go"}}, diff --git a/spindle/server.go b/spindle/server.go index 4855cf925..e80837485 100644 --- a/spindle/server.go +++ b/spindle/server.go @@ -28,6 +28,7 @@ import ( "tangled.org/core/gitutil" "tangled.org/core/idresolver" "tangled.org/core/jetstream" + "tangled.org/core/knotfeed" kgit "tangled.org/core/knotserver/git" "tangled.org/core/log" "tangled.org/core/notifier" @@ -300,11 +301,16 @@ func New(ctx context.Context, cfg *config.Config, d *db.DB, engines map[string]m return nil, fmt.Errorf("failed to start jetstream consumer: %w", err) } spindle.feed = feed.New(log.SubLogger(logger, "knotfeed"), feed.Hooks{ - LoadCursor: func(ctx context.Context, knot string) (int64, error) { - return spindle.db.LoadFeedCursor(knot) + LoadCursor: func(ctx context.Context, knot string) (knotfeed.Cursor, error) { + cursor, err := spindle.db.LoadFeedCursor(knot) + if errors.Is(err, knotfeed.ErrUnrecognizedFeed) { + logger.Warn("stored feed didn't parse, so resuming live", "knot", knot, "err", err) + return knotfeed.Cursor{}, nil + } + return cursor, err }, - StoreCursor: func(ctx context.Context, knot string, seq int64) error { - return spindle.db.StoreFeedCursor(knot, seq) + StoreCursor: func(ctx context.Context, knot string, cursor knotfeed.Cursor) error { + return spindle.db.StoreFeedCursor(knot, cursor) }, Handle: spindle.handleKnotFeed, OutdatedReplay: spindle.feedOutdatedReplay,