diff --git a/pkg/api/api_internal.go b/pkg/api/api_internal.go index 25f2f6b0..6bb2724c 100644 --- a/pkg/api/api_internal.go +++ b/pkg/api/api_internal.go @@ -115,6 +115,11 @@ func (a *StreamplaceAPI) InternalHandler(ctx context.Context) (http.Handler, err // store and attest to it via place.stream.media.origin. router.POST("/vod-transfer", a.HandleVODTransfer(ctx)) + // Rebuild the local media.origin index from our own server repo, for + // blobs we host but never indexed (a dropped firehose event, or a + // --secure node whose self-subscription never connected). + router.POST("/reindex-origins", a.HandleReindexOrigins(ctx)) + router.Handler("GET", "/metrics", promhttp.Handler()) // Legacy disk-served HLS (ffconcat -> latest.mp4 -> segment/:file, all reading diff --git a/pkg/api/reindex_origins.go b/pkg/api/reindex_origins.go new file mode 100644 index 00000000..e3cd7be4 --- /dev/null +++ b/pkg/api/reindex_origins.go @@ -0,0 +1,96 @@ +package api + +import ( + "context" + "encoding/json" + "net/http" + + "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. +// +// 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 +// federating a burst of churn. Safe to run while the node is live — every write +// is the same upsert the firehose would have done. +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 + } + + log.Log(reqCtx, "reindexed media origins from server repo", + "serverDid", 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 { + log.Error(reqCtx, "error writing reindex-origins response", "error", err) + } + } +} diff --git a/pkg/api/reindex_origins_test.go b/pkg/api/reindex_origins_test.go new file mode 100644 index 00000000..f457b4a1 --- /dev/null +++ b/pkg/api/reindex_origins_test.go @@ -0,0 +1,161 @@ +package api + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "testing" + + "github.com/bluesky-social/indigo/atproto/syntax" + "github.com/julienschmidt/httprouter" + "github.com/stretchr/testify/require" + + "stream.place/streamplace/pkg/atproto" + "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/statedb" +) + +func mustURI(t *testing.T, s string) syntax.ATURI { + t.Helper() + u, err := syntax.ParseATURI(s) + require.NoError(t, err) + return u +} + +// putHostedVideo writes a media.track backed by blobCID plus a sourceTracks +// video referencing it — the shape getVideoList resolves to a content blob. +func putHostedVideo(t *testing.T, m model.Model, videoURI, trackURI, blobCID string) { + t.Helper() + ctx := context.Background() + require.NoError(t, m.UpsertMediaTrack(ctx, 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", + }, + }, + }, mustURI(t, trackURI))) + + require.NoError(t, m.UpsertVideo(ctx, placestream.Video{ + LexiconTypeID: constants.PLACE_STREAM_VIDEO, + Title: videoURI, + 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: "bafytrackcid"}, + }, + }, + }, + }, mustURI(t, videoURI))) +} + +// TestReindexOriginsRepairsListing reproduces the production failure and its +// repair: the node has published media.origin records to its server repo and +// genuinely holds the blobs, but the firehose never round-tripped them into the +// local index (on a --secure node the self-subscription could not connect at +// all). The videos are therefore invisible to getVideoList despite being fully +// playable. Reindexing from the server repo — the authority — must restore them. +func TestReindexOriginsRepairsListing(t *testing.T) { + ctx := context.Background() + const serverHost = "server1.example.com" + serverDID := "did:web:" + serverHost + + cli := config.CLI{ + BroadcasterHost: "example.com", + ServerHost: serverHost, + DBURL: ":memory:", + } + cli.DataDir = t.TempDir() + + mod, err := model.MakeDB(":memory:") + require.NoError(t, err) + state, err := statedb.MakeDB(ctx, &cli, nil, mod) + require.NoError(t, err) + + handle, err := atproto.MakeServerRepo(ctx, &cli, state) + require.NoError(t, err) + defer 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 { + require.NoError(t, atproto.CommitServerRepoRecord(ctx, &cli, + constants.PLACE_STREAM_MEDIA_ORIGIN, blob, &placestream.MediaOrigin{ + LexiconTypeID: constants.PLACE_STREAM_MEDIA_ORIGIN, + Blob: blob, + Size: int64(1000 + i), + MimeType: "video/mp4", + })) + putHostedVideo(t, + mod, + fmt.Sprintf("at://did:plc:alice/place.stream.video/v%d", i), + fmt.Sprintf("at://did:plc:alice/place.stream.media.track/t%d", i), + blob, + ) + } + + // The symptom: nothing is listable, even though every blob is ours. + before, err := mod.GetVideoList(ctx, "", 25, "", serverDID) + require.NoError(t, err) + require.Empty(t, before.Videos, "precondition: origins are unindexed, so nothing lists") + + // 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, "", "") + require.NoError(t, err) + require.Len(t, unfiltered.Videos, len(blobs)) + + 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 + require.NoError(t, json.NewDecoder(rr.Body).Decode(&res)) + require.Equal(t, serverDID, res.ServerDID) + require.Equal(t, len(blobs), res.Scanned) + require.Equal(t, len(blobs), res.Indexed) + require.Empty(t, res.Errors) + + // 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)) + + // 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")) + require.NoError(t, err) + require.Equal(t, "blobBBB", origin.Blob) + require.Equal(t, int64(1001), origin.Size) + require.Equal(t, "video/mp4", origin.MimeType) + + // Idempotent: a second pass writes the same rows and changes nothing. + 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 + 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)) +} diff --git a/pkg/api/vod_transfer.go b/pkg/api/vod_transfer.go index 49c54beb..21775eda 100644 --- a/pkg/api/vod_transfer.go +++ b/pkg/api/vod_transfer.go @@ -11,10 +11,8 @@ import ( "github.com/julienschmidt/httprouter" "stream.place/streamplace/pkg/comatproto" - "stream.place/streamplace/pkg/constants" "stream.place/streamplace/pkg/errors" "stream.place/streamplace/pkg/log" - "stream.place/streamplace/pkg/placestream" "stream.place/streamplace/pkg/vod" ) @@ -117,19 +115,7 @@ func (a *StreamplaceAPI) HandleVODTransfer(ctx context.Context) httprouter.Handl // back. The rkey is the blob CID by convention, and the authority is our // ServerDID — the same (server_did, blob) key getVideoList filters on. func (a *StreamplaceAPI) indexOwnMediaOrigin(ctx context.Context, contentCID string, size int64) error { - aturi, err := syntax.ParseATURI(fmt.Sprintf( - "at://%s/%s/%s", a.CLI.ServerDID(), constants.PLACE_STREAM_MEDIA_ORIGIN, contentCID, - )) - if err != nil { - return fmt.Errorf("build origin uri: %w", err) - } - rec := placestream.MediaOrigin{ - LexiconTypeID: constants.PLACE_STREAM_MEDIA_ORIGIN, - Blob: contentCID, - Size: size, - MimeType: "video/mp4", - } - return a.Model.UpsertMediaOrigin(ctx, rec, aturi) + return a.Model.UpsertOwnMediaOrigin(ctx, a.CLI.ServerDID(), contentCID, size, "video/mp4") } // resolveVideoContentBlob walks a place.stream.video record in the local diff --git a/pkg/model/media_origin.go b/pkg/model/media_origin.go index b811f73e..c3fff08b 100644 --- a/pkg/model/media_origin.go +++ b/pkg/model/media_origin.go @@ -11,6 +11,7 @@ import ( glex "github.com/streamplace/glex/runtime" "gorm.io/gorm" "stream.place/streamplace/pkg/aqtime" + "stream.place/streamplace/pkg/constants" "stream.place/streamplace/pkg/placestream" "stream.place/streamplace/pkg/spid" ) @@ -63,6 +64,32 @@ func (m *DBModel) UpsertMediaOrigin(ctx context.Context, rec placestream.MediaOr return m.DB.WithContext(ctx).Save(o).Error } +// UpsertOwnMediaOrigin indexes this node's own attestation that it holds a +// blob, building the record and AT-URI the way the publisher does: rkey is the +// blob CID by convention, and the authority is our ServerDID — the same +// (server_did, blob) key GetVideoList filters on. +// +// Publishing the record to the server repo and indexing it locally are separate +// steps, normally bridged by the record federating back to us over the firehose. +// That round-trip is a single point of failure, and when it drops an event the +// video plays fine by direct link but never appears in any listing, with nothing +// to reconcile it afterward. Callers that publish an origin should index it here +// too; both paths are idempotent, so whichever lands second is a no-op. +func (m *DBModel) UpsertOwnMediaOrigin(ctx context.Context, serverDID, blobCID string, size int64, mimeType string) error { + aturi, err := syntax.ParseATURI(fmt.Sprintf( + "at://%s/%s/%s", serverDID, constants.PLACE_STREAM_MEDIA_ORIGIN, blobCID, + )) + if err != nil { + return fmt.Errorf("build media origin uri: %w", err) + } + return m.UpsertMediaOrigin(ctx, placestream.MediaOrigin{ + LexiconTypeID: constants.PLACE_STREAM_MEDIA_ORIGIN, + Blob: blobCID, + Size: size, + MimeType: mimeType, + }, aturi) +} + func (m *DBModel) DeleteMediaOrigin(ctx context.Context, uri string) error { return m.DB.WithContext(ctx).Where("uri = ?", uri).Delete(&MediaOrigin{}).Error } diff --git a/pkg/model/model.go b/pkg/model/model.go index ad23ff8e..35109abf 100644 --- a/pkg/model/model.go +++ b/pkg/model/model.go @@ -148,6 +148,7 @@ type Model interface { GetMediaTracksByBlob(ctx context.Context, blob string) ([]*MediaTrack, error) UpsertMediaOrigin(ctx context.Context, rec placestream.MediaOrigin, aturi syntax.ATURI) error + UpsertOwnMediaOrigin(ctx context.Context, serverDID, blobCID string, size int64, mimeType string) error DeleteMediaOrigin(ctx context.Context, uri string) error GetMediaOriginByURI(ctx context.Context, uri string) (placestream.MediaOrigin, error) GetMediaOriginsByBlob(ctx context.Context, blob string) ([]*MediaOrigin, error) diff --git a/pkg/statedb/media_origin.go b/pkg/statedb/media_origin.go new file mode 100644 index 00000000..99ae4b05 --- /dev/null +++ b/pkg/statedb/media_origin.go @@ -0,0 +1,32 @@ +package statedb + +import ( + "context" + "fmt" +) + +// IndexOwnMediaOrigin records in the local index that this node holds the given +// blob, so a freshly published VOD is listable immediately. +// +// pkg/vod publishes the place.stream.media.origin record to the server repo and +// would otherwise wait for it to federate back over the firehose before the +// index learns about it — a round-trip that has silently failed for extended +// stretches (it is down entirely on a --secure node whose self-subscription +// can't dial its own listener). getVideoList hides any video without an origin +// row, so a dropped event means a permanently unlistable-but-playable video. +// +// This lives on StatefulDB because pkg/vod deliberately doesn't import +// pkg/model (it can then run as a standalone microservice), but it already +// holds a *StatefulDB — so this is the seam that reaches the index without +// widening that dependency. +func (state *StatefulDB) IndexOwnMediaOrigin(ctx context.Context, blobCID string, size int64, mimeType string) error { + if state.model == nil { + // Standalone/microservice deployments run without an index; the + // firehose path on the indexing node is then the only writer. + return nil + } + if err := state.model.UpsertOwnMediaOrigin(ctx, state.CLI.ServerDID(), blobCID, size, mimeType); err != nil { + return fmt.Errorf("index own media origin: %w", err) + } + return nil +} diff --git a/pkg/statedb/media_origin_test.go b/pkg/statedb/media_origin_test.go new file mode 100644 index 00000000..708ab662 --- /dev/null +++ b/pkg/statedb/media_origin_test.go @@ -0,0 +1,65 @@ +package statedb + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + + "stream.place/streamplace/pkg/config" + "stream.place/streamplace/pkg/constants" + "stream.place/streamplace/pkg/model" +) + +// TestIndexOwnMediaOrigin checks the seam pkg/vod uses to index an origin +// without waiting on the firehose. The authority must be our ServerDID — that +// is the exact (server_did, blob) key getVideoList filters on, so getting it +// wrong reintroduces the silent invisibility this exists to prevent. +func TestIndexOwnMediaOrigin(t *testing.T) { + ctx := context.Background() + cli := config.CLI{ + BroadcasterHost: "example.com", + ServerHost: "server1.example.com", + DBURL: ":memory:", + } + cli.DataDir = t.TempDir() + + mod, err := model.MakeDB(":memory:") + require.NoError(t, err) + state, err := MakeDB(ctx, &cli, nil, mod) + require.NoError(t, err) + + require.NoError(t, state.IndexOwnMediaOrigin(ctx, "blobXYZ", 4096, "video/mp4")) + + origin, err := mod.GetMediaOriginByURI(ctx, + "at://did:web:server1.example.com/"+constants.PLACE_STREAM_MEDIA_ORIGIN+"/blobXYZ") + require.NoError(t, err) + require.Equal(t, "blobXYZ", origin.Blob) + require.Equal(t, int64(4096), origin.Size) + require.Equal(t, "video/mp4", origin.MimeType) + + // It is the row getVideoList looks for, under our DID and no other. + hosted, err := mod.GetVideoList(ctx, "", 25, "", "did:web:server1.example.com") + require.NoError(t, err) + require.Empty(t, hosted.Videos) // no video records seeded; the point is it doesn't error + + // Idempotent: publishing the same origin twice (retry, or the firehose + // copy landing afterward) must not duplicate or fail. + require.NoError(t, state.IndexOwnMediaOrigin(ctx, "blobXYZ", 4096, "video/mp4")) + + origins, err := mod.GetMediaOriginsByBlob(ctx, "blobXYZ") + require.NoError(t, err) + require.Len(t, origins, 1) +} + +// TestIndexOwnMediaOriginNoModel covers the standalone/microservice case: with +// no index attached there is nothing to write, and that must not be an error. +func TestIndexOwnMediaOriginNoModel(t *testing.T) { + ctx := context.Background() + cli := config.CLI{ServerHost: "server1.example.com", DBURL: ":memory:"} + cli.DataDir = t.TempDir() + + state, err := MakeDB(ctx, &cli, nil, nil) + require.NoError(t, err) + require.NoError(t, state.IndexOwnMediaOrigin(ctx, "blobXYZ", 1, "video/mp4")) +} diff --git a/pkg/vod/publish.go b/pkg/vod/publish.go index de39a63e..0e403e2a 100644 --- a/pkg/vod/publish.go +++ b/pkg/vod/publish.go @@ -74,6 +74,19 @@ func publishRecords(ctx context.Context, p publishParams) error { return fmt.Errorf("publish origin: %w", err) } + // The commit above reaches the local index by federating back to us over + // the firehose, which is a single point of failure: miss that one event and + // the video plays fine by direct link but never shows up in any listing, + // with nothing to reconcile it later. Index it directly as well. Not fatal + // — a VOD that published its records but lost a race with the indexer is + // still a successful VOD, and the firehose upsert (or a reindex) converges + // it. Idempotent either way, keyed by URI. + if p.state != nil { + if err := p.state.IndexOwnMediaOrigin(ctx, p.cid, p.size, p.mimeType); err != nil { + log.Error(ctx, "failed to index own media.origin locally", "cid", p.cid, "error", err) + } + } + // Serialize the probe so publishDraft can publish the track records later // without re-probing the blob. probeJSON, err := marshalProbe(p.probe)