From ccc907c5046916fbfcff4e80a2001e785f44a896 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Wed, 30 Sep 2026 16:25:31 -0700 Subject: [PATCH] fix: recover historical VOD origins before serving listings --- .maestro/logged-out/vod.yaml | 29 ++++ .../installing/downloading-streamplace.md | 17 ++ js/e2e-web/flows/vod.spec.ts | 26 +++ pkg/api/reindex_origins.go | 73 ++------- pkg/api/reindex_origins_test.go | 57 +++++-- pkg/atproto/media_origins.go | 52 ++++++ pkg/atproto/server_repo.go | 8 +- pkg/cmd/e2e.go | 4 + pkg/cmd/e2e_recovery.go | 148 ++++++++++++++++++ pkg/cmd/streamplace.go | 13 ++ 10 files changed, 348 insertions(+), 79 deletions(-) create mode 100644 .maestro/logged-out/vod.yaml create mode 100644 pkg/atproto/media_origins.go create mode 100644 pkg/cmd/e2e_recovery.go diff --git a/.maestro/logged-out/vod.yaml b/.maestro/logged-out/vod.yaml new file mode 100644 index 000000000..a84228da6 --- /dev/null +++ b/.maestro/logged-out/vod.yaml @@ -0,0 +1,29 @@ +appId: ${APP_ID} +--- +# This VOD was indexed before node launch, with its media.origin only in the +# server repo. It must be recovered at startup, not by a test-triggered reindex. +- launchApp +- tapOn: + text: "Videos" +# iOS can collapse the card's title and author into its link accessibility +# label; substring matching works for that and Android's separate title text. +- scrollUntilVisible: + element: + text: ".*e2e recovered video.*" + direction: DOWN + timeout: 30000 +- assertNotVisible: + text: ".*e2e unhosted video.*" +- takeScreenshot: 08-01-recovered-listing +- tapOn: + text: ".*e2e recovered video.*" +- waitForAnimationToEnd +# Require the detail screen, not just the card that was tapped. The Videos +# tab/header disappears, and the VOD metadata renders the exact record title. +- assertNotVisible: + text: "Videos" +- extendedWaitUntil: + visible: + text: "e2e recovered video" + timeout: 30000 +- takeScreenshot: 08-02-recovered-video diff --git a/js/docs/src/content/docs/guides/installing/downloading-streamplace.md b/js/docs/src/content/docs/guides/installing/downloading-streamplace.md index 412593408..cbce6019a 100644 --- a/js/docs/src/content/docs/guides/installing/downloading-streamplace.md +++ b/js/docs/src/content/docs/guides/installing/downloading-streamplace.md @@ -79,3 +79,20 @@ host networking. So that command would look something like: Binaries for all platforms are available to download from [our GitLab server](https://git.stream.place/streamplace/streamplace/-/releases). + +## Recovering missing video listings + +Video listings include only videos whose content blob has a `place.stream.media.origin` +indexed under this node's server DID. A missed origin firehose event can therefore +hide a VOD from listings even when it still plays by direct link. + +Streamplace rebuilds its own origin index from its durable server repository at +startup, before serving requests. After upgrading, restart the node to repair +historical gaps, including origins older than the 72-hour commit replay window. +This does not republish records or add videos hosted only by other nodes. + +For a live node, the internal admin listener also supports `POST /reindex-origins`. +Its JSON response reports `serverDid`, `scanned`, `indexed`, and any per-record +`errors`. Use the configured internal address locally; do not expose this listener +publicly. Startup fails if reconciliation cannot complete, rather than silently +serving an incomplete hosting index. diff --git a/js/e2e-web/flows/vod.spec.ts b/js/e2e-web/flows/vod.spec.ts index d77bc5d96..4adeeb839 100644 --- a/js/e2e-web/flows/vod.spec.ts +++ b/js/e2e-web/flows/vod.spec.ts @@ -6,6 +6,32 @@ import { expect, test } from "@playwright/test"; // has loaded, so open tabs and bookmarks tell videos apart. const VIDEO_URI = process.env.VIDEO_URI; const VIDEO_TITLE = "e2e test video"; +const RECOVERED_VIDEO_TITLE = "e2e recovered video"; +const UNHOSTED_VIDEO_TITLE = "e2e unhosted video"; + +for (const scope of ["global", "profile"] as const) { + test(`08-vod: ${scope} listing recovers hosted videos after boot`, async ({ + page, + }) => { + const handle = process.env.ACCOUNT_HANDLE; + if (scope === "profile" && !handle) { + throw new Error( + "ACCOUNT_HANDLE is required for the profile video listing", + ); + } + + // This fixture's video and track were indexed before launch, but its own + // media.origin existed only in the server repo. The real listing API must + // expose it after startup reconciliation, without exposing unhosted videos. + await page.goto(scope === "global" ? "/video" : `/${handle}/video`); + await expect( + page.getByText(RECOVERED_VIDEO_TITLE, { exact: true }), + ).toBeVisible({ timeout: 30_000 }); + await expect( + page.getByText(UNHOSTED_VIDEO_TITLE, { exact: true }), + ).toHaveCount(0); + }); +} function videoPath(): string { const match = VIDEO_URI?.match( diff --git a/pkg/api/reindex_origins.go b/pkg/api/reindex_origins.go index e3cd7be4e..8a32fcd37 100644 --- a/pkg/api/reindex_origins.go +++ b/pkg/api/reindex_origins.go @@ -7,38 +7,16 @@ import ( "github.com/julienschmidt/httprouter" "stream.place/streamplace/pkg/atproto" - "stream.place/streamplace/pkg/constants" "stream.place/streamplace/pkg/errors" "stream.place/streamplace/pkg/log" - "stream.place/streamplace/pkg/placestream" ) -// reindexOriginsPageSize is how many origin records we pull per repo page. -// ServerRepoListRecords takes the server repo lock and decodes each record -// body, so we page rather than asking for everything at once. -const reindexOriginsPageSize = 100 - -// reindexOriginsResponse reports what a reindex pass did. Scanned counts the -// records walked in the server repo; Indexed counts the rows written. They -// differ only when a record fails to decode or upsert, which is what Errors -// enumerates. -type reindexOriginsResponse struct { - ServerDID string `json:"serverDid"` - Scanned int `json:"scanned"` - Indexed int `json:"indexed"` - Errors []string `json:"errors,omitempty"` -} - // HandleReindexOrigins rebuilds the local place.stream.media.origin index from // this node's own server repo. // -// The server repo is the authority on what we host, and it is always complete: -// publishOrigin commits there synchronously. The local index is the derived, -// lossy copy — it is only ever written by the firehose sync path, so any event -// that connection drops is gone for good, and getVideoList then hides a video -// that the node can in fact serve. (On a --secure node the self-subscription -// could not dial its own listener at all, so nothing published this way was -// ever indexed.) This walks the authority and replays it into the index. +// The server repo is authoritative; the local hosting index may be incomplete +// after a missed firehose event. Startup runs the same reconciliation before +// serving listings. This endpoint also allows repair while the node is live. // // Idempotent, and cheap in the ways that matter: it writes no repo commits and // emits no firehose events, so it can be run repeatedly and on any node without @@ -47,46 +25,17 @@ type reindexOriginsResponse struct { func (a *StreamplaceAPI) HandleReindexOrigins(ctx context.Context) httprouter.Handle { return func(w http.ResponseWriter, r *http.Request, p httprouter.Params) { reqCtx := r.Context() - serverDID := a.CLI.ServerDID() - res := reindexOriginsResponse{ServerDID: serverDID} - - cursor := "" - for { - page, err := atproto.ServerRepoListRecords( - reqCtx, constants.PLACE_STREAM_MEDIA_ORIGIN, cursor, - reindexOriginsPageSize, serverDID, nil, - ) - if err != nil { - errors.WriteHTTPInternalServerError(w, "list server repo origins", err) - return - } - for _, rec := range page.Records { - res.Scanned++ - origin, ok := rec.Value.Val.(*placestream.MediaOrigin) - if !ok { - res.Errors = append(res.Errors, rec.Uri+": not a media.origin record") - continue - } - // Index under the blob the record names rather than the rkey. - // They agree by convention, but the record is the data and the - // rkey is only a naming convention, so trust the record. - if err := a.Model.UpsertOwnMediaOrigin( - reqCtx, serverDID, origin.Blob, origin.Size, origin.MimeType, - ); err != nil { - log.Error(reqCtx, "reindex origins: upsert failed", "uri", rec.Uri, "error", err) - res.Errors = append(res.Errors, rec.Uri+": "+err.Error()) - continue - } - res.Indexed++ - } - if page.Cursor == nil || *page.Cursor == "" { - break - } - cursor = *page.Cursor + res, err := atproto.ReindexOwnMediaOrigins(reqCtx, a.Model, a.CLI.ServerDID()) + if err != nil { + errors.WriteHTTPInternalServerError(w, "reindex origins", err) + return + } + for _, failure := range res.Errors { + log.Error(reqCtx, "reindex origins: record failed", "error", failure) } log.Log(reqCtx, "reindexed media origins from server repo", - "serverDid", serverDID, "scanned", res.Scanned, "indexed", res.Indexed, "errors", len(res.Errors)) + "serverDid", res.ServerDID, "scanned", res.Scanned, "indexed", res.Indexed, "errors", len(res.Errors)) w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(res); err != nil { diff --git a/pkg/api/reindex_origins_test.go b/pkg/api/reindex_origins_test.go index f457b4a1c..0cee2d8c0 100644 --- a/pkg/api/reindex_origins_test.go +++ b/pkg/api/reindex_origins_test.go @@ -84,18 +84,30 @@ func TestReindexOriginsRepairsListing(t *testing.T) { handle, err := atproto.MakeServerRepo(ctx, &cli, state) require.NoError(t, err) - defer handle.Close() + defer func() { require.NoError(t, handle.Close()) }() t.Cleanup(func() { atproto.ServerRepo = nil atproto.ServerCarStore = nil atproto.ServerPubMultibase = "" }) - // Three VODs this node hosts: origin committed to the server repo (as - // publishOrigin does), video+track indexed (as the user's own firehose - // events did), but no media_origins row — the dropped half. - blobs := []string{"blobAAA", "blobBBB", "blobCCC"} - for i, blob := range blobs { + // More than one origin page, persisted without local hosting rows. + // An unrelated collection must not interfere with origin reconciliation. + updatedAt := "2026-07-25T00:00:00Z" + require.NoError(t, atproto.CommitServerRepoRecord(ctx, &cli, + constants.PLACE_STREAM_LIVE_VIEWERCOUNT, "did:plc:alice", &placestream.LiveViewerCount{ + LexiconTypeID: constants.PLACE_STREAM_LIVE_VIEWERCOUNT, + Count: 1, + Server: serverDID, + Streamer: "did:plc:alice", + UpdatedAt: &updatedAt, + })) + blobs := make([]string, 103) + wantVideos := make(map[string]bool, len(blobs)) + for i := range blobs { + blob := fmt.Sprintf("blob%03d", i) + blobs[i] = blob + wantVideos[fmt.Sprintf("at://did:plc:alice/place.stream.video/v%d", i)] = true require.NoError(t, atproto.CommitServerRepoRecord(ctx, &cli, constants.PLACE_STREAM_MEDIA_ORIGIN, blob, &placestream.MediaOrigin{ LexiconTypeID: constants.PLACE_STREAM_MEDIA_ORIGIN, @@ -110,6 +122,9 @@ func TestReindexOriginsRepairsListing(t *testing.T) { blob, ) } + require.NoError(t, handle.Close()) + handle, err = atproto.MakeServerRepo(ctx, &cli, state) + require.NoError(t, err) // The symptom: nothing is listable, even though every blob is ours. before, err := mod.GetVideoList(ctx, "", 25, "", serverDID) @@ -118,16 +133,16 @@ func TestReindexOriginsRepairsListing(t *testing.T) { // Unfiltered the videos are plainly there — they are hidden by the hosted // filter alone, which is exactly why they still play by direct link. - unfiltered, err := mod.GetVideoList(ctx, "", 25, "", "") + unfiltered, err := mod.GetVideoList(ctx, "", 100, "", "") require.NoError(t, err) - require.Len(t, unfiltered.Videos, len(blobs)) + require.Len(t, unfiltered.Videos, 100) a := StreamplaceAPI{CLI: &cli, Model: mod} rr := httptest.NewRecorder() a.HandleReindexOrigins(ctx)(rr, httptest.NewRequest(http.MethodPost, "/reindex-origins", nil), httprouter.Params{}) require.Equal(t, http.StatusOK, rr.Result().StatusCode) - var res reindexOriginsResponse + var res atproto.MediaOriginReindexResult require.NoError(t, json.NewDecoder(rr.Body).Decode(&res)) require.Equal(t, serverDID, res.ServerDID) require.Equal(t, len(blobs), res.Scanned) @@ -137,13 +152,27 @@ func TestReindexOriginsRepairsListing(t *testing.T) { // The repair: every video the node hosts is listable again. after, err := mod.GetVideoList(ctx, "", 25, "", serverDID) require.NoError(t, err) - require.Len(t, after.Videos, len(blobs)) + gotVideos := make(map[string]bool, len(blobs)) + cursor := "" + for { + page, err := mod.GetVideoList(ctx, "", 100, cursor, serverDID) + require.NoError(t, err) + for _, video := range page.Videos { + require.False(t, gotVideos[video.Uri], "duplicate listing entry") + gotVideos[video.Uri] = true + } + if page.Cursor == nil { + break + } + cursor = *page.Cursor + } + require.Equal(t, wantVideos, gotVideos) // Size/mimeType came from the record body, not invented from the rkey. origin, err := mod.GetMediaOriginByURI(ctx, fmt.Sprintf( - "at://%s/%s/%s", serverDID, constants.PLACE_STREAM_MEDIA_ORIGIN, "blobBBB")) + "at://%s/%s/%s", serverDID, constants.PLACE_STREAM_MEDIA_ORIGIN, "blob001")) require.NoError(t, err) - require.Equal(t, "blobBBB", origin.Blob) + require.Equal(t, "blob001", origin.Blob) require.Equal(t, int64(1001), origin.Size) require.Equal(t, "video/mp4", origin.MimeType) @@ -151,11 +180,11 @@ func TestReindexOriginsRepairsListing(t *testing.T) { rr2 := httptest.NewRecorder() a.HandleReindexOrigins(ctx)(rr2, httptest.NewRequest(http.MethodPost, "/reindex-origins", nil), httprouter.Params{}) require.Equal(t, http.StatusOK, rr2.Result().StatusCode) - var res2 reindexOriginsResponse + var res2 atproto.MediaOriginReindexResult require.NoError(t, json.NewDecoder(rr2.Body).Decode(&res2)) require.Equal(t, len(blobs), res2.Indexed) again, err := mod.GetVideoList(ctx, "", 25, "", serverDID) require.NoError(t, err) - require.Len(t, again.Videos, len(blobs)) + require.Equal(t, after.Videos, again.Videos) } diff --git a/pkg/atproto/media_origins.go b/pkg/atproto/media_origins.go new file mode 100644 index 000000000..f109931e4 --- /dev/null +++ b/pkg/atproto/media_origins.go @@ -0,0 +1,52 @@ +package atproto + +import ( + "context" + "fmt" + + "stream.place/streamplace/pkg/constants" + "stream.place/streamplace/pkg/model" + "stream.place/streamplace/pkg/placestream" +) + +// MediaOriginReindexResult reports the records read from the server repo and +// written to the local index. Errors contains individual record/upsert failures. +type MediaOriginReindexResult struct { + ServerDID string `json:"serverDid"` + Scanned int `json:"scanned"` + Indexed int `json:"indexed"` + Errors []string `json:"errors,omitempty"` +} + +// ReindexOwnMediaOrigins repairs the derived hosting index from this node's +// authoritative server repo. Run before serving listings, since old origins +// may predate direct local indexing or have missed their firehose round-trip. +// It writes no repo commits and leaves other nodes' origin rows untouched. +func ReindexOwnMediaOrigins(ctx context.Context, mod model.Model, serverDID string) (MediaOriginReindexResult, error) { + res := MediaOriginReindexResult{ServerDID: serverDID} + cursor := "" + for { + page, err := ServerRepoListRecords(ctx, constants.PLACE_STREAM_MEDIA_ORIGIN, cursor, 100, serverDID, nil) + if err != nil { + return res, fmt.Errorf("list server repo origins: %w", err) + } + for _, rec := range page.Records { + res.Scanned++ + origin, ok := rec.Value.Val.(*placestream.MediaOrigin) + if !ok { + res.Errors = append(res.Errors, rec.Uri+": not a media.origin record") + continue + } + // The record's blob is authoritative; its rkey is only a convention. + if err := mod.UpsertOwnMediaOrigin(ctx, serverDID, origin.Blob, origin.Size, origin.MimeType); err != nil { + res.Errors = append(res.Errors, rec.Uri+": "+err.Error()) + continue + } + res.Indexed++ + } + if page.Cursor == nil || *page.Cursor == "" { + return res, nil + } + cursor = *page.Cursor + } +} diff --git a/pkg/atproto/server_repo.go b/pkg/atproto/server_repo.go index 215eb866c..3e28cede5 100644 --- a/pkg/atproto/server_repo.go +++ b/pkg/atproto/server_repo.go @@ -3,6 +3,7 @@ package atproto import ( "bytes" "context" + "database/sql" "errors" "fmt" "os" @@ -109,10 +110,11 @@ type ServerCommitEvent struct { type serverRepoCloser struct { carStore *carstore.SQLiteStore + commitDB *sql.DB } func (c *serverRepoCloser) Close() error { - return c.carStore.Close() + return errors.Join(c.carStore.Close(), c.commitDB.Close()) } func MakeServerRepo(ctx context.Context, cli *config.CLI, state *statedb.StatefulDB) (Closer, error) { @@ -205,7 +207,7 @@ func MakeServerRepo(ctx context.Context, cli *config.CLI, state *statedb.Statefu // Open local SQLite for commit events commitDBPath := cli.DataFilePath([]string{"server-repo", "commits.db"}) - db, err := gorm.Open(sqlite.Open(commitDBPath), &gorm.Config{}) + db, err := gorm.Open(sqlite.Open(commitDBPath), &gorm.Config{Logger: config.GormLogger}) if err != nil { return nil, fmt.Errorf("failed to open server commit db at %s: %w", commitDBPath, err) } @@ -226,7 +228,7 @@ func MakeServerRepo(ctx context.Context, cli *config.CLI, state *statedb.Statefu log.Warn(ctx, "failed to prune old server commit events", "error", err) } - return &serverRepoCloser{carStore: sqliteStore}, nil + return &serverRepoCloser{carStore: sqliteStore, commitDB: sqlDB}, nil } func OpenServerRepo(ctx context.Context) (*atrepo.Repo, *carstore.DeltaSession, error) { diff --git a/pkg/cmd/e2e.go b/pkg/cmd/e2e.go index 44f575720..c2f335c33 100644 --- a/pkg/cmd/e2e.go +++ b/pkg/cmd/e2e.go @@ -254,6 +254,10 @@ func runE2E(ctx context.Context, devEnvPath, httpsPDSHost, httpsStationHost stri } defer os.RemoveAll(dataDir) //nolint:errcheck + if err := seedE2ERecovery(ctx, dataDir, broadcasterHost, out.Did, xrpcc); err != nil { + return fmt.Errorf("seed startup recovery fixtures: %w", err) + } + nodeCmd := exec.CommandContext(ctx, self) // Inherit the parent environment (dev builds need LD_LIBRARY_PATH etc.) // but strip any SP_ vars so the node only gets our config. diff --git a/pkg/cmd/e2e_recovery.go b/pkg/cmd/e2e_recovery.go new file mode 100644 index 000000000..833215499 --- /dev/null +++ b/pkg/cmd/e2e_recovery.go @@ -0,0 +1,148 @@ +package cmd + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/bluesky-social/indigo/atproto/syntax" + "github.com/bluesky-social/indigo/util" + "github.com/bluesky-social/indigo/xrpc" + "gorm.io/driver/sqlite" + "gorm.io/gorm" + "stream.place/streamplace/pkg/atproto" + "stream.place/streamplace/pkg/bdasl" + "stream.place/streamplace/pkg/comatproto" + "stream.place/streamplace/pkg/config" + "stream.place/streamplace/pkg/constants" + "stream.place/streamplace/pkg/model" + "stream.place/streamplace/pkg/placestream" + "stream.place/streamplace/pkg/spid" + "stream.place/streamplace/pkg/statedb" +) + +// seedE2ERecovery reproduces an old node's persisted index and server repo. +// These are listing fixtures only: no media payload is written or promised. +func seedE2ERecovery(ctx context.Context, dataDir, broadcasterHost, accountDID string, client *xrpc.Client) (err error) { + // Match the fork's defaults, including ServerHost's broadcaster fallback + // and the host-scoped server signing key stored in state.sqlite. + cli := config.CLI{ + DataDir: dataDir, + BroadcasterHost: broadcasterHost, + ServerHost: broadcasterHost, + DBURL: "sqlite://" + dataDir + "/state.sqlite", + } + mod, err := model.MakeDB(cli.DataFilePath([]string{"index"})) + if err != nil { + return fmt.Errorf("open recovery index: %w", err) + } + indexDB, err := mod.(*model.DBModel).DB.DB() + if err != nil { + return err + } + defer func() { err = errors.Join(err, indexDB.Close()) }() + state, err := statedb.MakeDB(ctx, &cli, nil, mod) + if err != nil { + return fmt.Errorf("open recovery state: %w", err) + } + stateDB, err := state.DB.DB() + if err != nil { + return err + } + defer func() { err = errors.Join(err, stateDB.Close()) }() + serverRepo, err := atproto.MakeServerRepo(ctx, &cli, state) + if err != nil { + return fmt.Errorf("open recovery server repo: %w", err) + } + // All SQLite handles are closed on return, before nodeCmd.Start opens + // the same files. The temporary data directory remains harness-owned. + defer func() { err = errors.Join(err, serverRepo.Close()) }() + + for _, title := range []string{"e2e recovered video", "e2e unhosted video"} { + blobCID := bdasl.CID([]byte(title)) + track := placestream.MediaTrack{ + LexiconTypeID: constants.PLACE_STREAM_MEDIA_TRACK, + Track: placestream.MediaTrack_Track{ + MediaDefs_MuxlTrack: &placestream.MediaDefs_MuxlTrack{ + LexiconTypeID: "place.stream.media.defs#muxlTrack", + Blob: blobCID, + TrackId: "1", + MediaType: "video", + }, + }, + } + // Publish the user's records too, so repo sweeps cannot delete the + // preindexed fixtures as records absent from the authoritative PDS. + trackURI, err := createRecord(ctx, client, constants.PLACE_STREAM_MEDIA_TRACK, accountDID, &track) + if err != nil { + return fmt.Errorf("publish recovery track: %w", err) + } + trackATURI, err := syntax.ParseATURI(trackURI) + if err != nil { + return err + } + if err := mod.UpsertMediaTrack(ctx, track, trackATURI); err != nil { + return err + } + trackCID, err := spid.GetCID(&track) + if err != nil { + return err + } + video := placestream.Video{ + LexiconTypeID: constants.PLACE_STREAM_VIDEO, + CreatedAt: time.Now().UTC().Format(util.ISO8601), + Title: title, + DurationMs: 10_000, + Source: placestream.Video_Source{ + MediaDefs_SourceTracks: &placestream.MediaDefs_SourceTracks{ + LexiconTypeID: "place.stream.media.defs#sourceTracks", + Tracks: []comatproto.RepoStrongRef{ + {LexiconTypeID: "com.atproto.repo.strongRef", Uri: trackURI, Cid: trackCID.String()}, + }, + }, + }, + } + videoURI, err := createRecord(ctx, client, constants.PLACE_STREAM_VIDEO, accountDID, &video) + if err != nil { + return fmt.Errorf("publish recovery video: %w", err) + } + videoATURI, err := syntax.ParseATURI(videoURI) + if err != nil { + return err + } + if err := mod.UpsertVideo(ctx, video, videoATURI); err != nil { + return err + } + if title == "e2e recovered video" { + // Deliberately do not index this origin. Startup must recover it + // from the node-owned repo, not an admin request or firehose. + if err := atproto.CommitServerRepoRecord(ctx, &cli, constants.PLACE_STREAM_MEDIA_ORIGIN, blobCID, &placestream.MediaOrigin{ + LexiconTypeID: constants.PLACE_STREAM_MEDIA_ORIGIN, + Blob: blobCID, + Size: int64(len(title)), + MimeType: "video/mp4", + }); err != nil { + return fmt.Errorf("commit recovery origin: %w", err) + } + } + } + // Age replay events past the node's 72-hour retention cutoff. The + // durable repo still owns the origin, but boot prunes its event so a + // self-subscription cannot mask a missing startup reconciliation. + commits, err := gorm.Open(sqlite.Open(cli.DataFilePath([]string{"server-repo", "commits.db"})), &gorm.Config{Logger: config.GormLogger}) + if err != nil { + return fmt.Errorf("open recovery commit events: %w", err) + } + commitDB, err := commits.DB() + if err != nil { + return fmt.Errorf("get recovery commit database: %w", err) + } + defer func() { err = errors.Join(err, commitDB.Close()) }() + if err := commits.WithContext(ctx).Model(&atproto.ServerCommitEvent{}). + Where("repo_did = ?", cli.ServerDID()). + Update("timestamp", time.Now().Add(-96*time.Hour)).Error; err != nil { + return fmt.Errorf("age recovery commit events: %w", err) + } + return nil +} diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go index 74bb64402..75427601b 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -256,6 +256,19 @@ func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFu } defer serverHandle.Close() + // Historical origin commits may never have reached the derived index. + // Repair them before accepting listing requests, without relying on the + // firehose's bounded commit history or a manual admin repair. + origins, err := atproto.ReindexOwnMediaOrigins(ctx, mod, cli.ServerDID()) + if err != nil { + return fmt.Errorf("reindex own media origins: %w", err) + } + if len(origins.Errors) > 0 { + return fmt.Errorf("reindex own media origins: %s", strings.Join(origins.Errors, "; ")) + } + log.Log(ctx, "reindexed own media origins at startup", + "serverDid", origins.ServerDID, "scanned", origins.Scanned, "indexed", origins.Indexed) + jwk, err := state.EnsureJWK(ctx, "jwk") if err != nil { return err -- knotmirror 0.0.1