From b32bd7af85e97e177ac8367b68583cd836ba9873 Mon Sep 17 00:00:00 2001 From: Dylan Jeffers Date: Thu, 3 Sep 2026 14:41:01 -0700 Subject: [PATCH 1/2] fix(api): a track with no track_cid is not streamable An upload whose track_cid never made it onto the track entity has audio sitting on the content node that no reader can address. The stream link was already left nil for these rows, and /stream already 404s, but is_streamable kept reporting true - so every client believed the track was healthy. The player spins on a dead URL, and mobile's share-to-story feeds that URL to ffmpeg, which fails with a generic "Sorry, something went wrong" instead of saying the track has no audio. Split the predicate in two. IsAudioAllowed keeps the old meaning - the track is not deleted and its owner is still active - and gates downloads and previews. IsStreamable now also requires a cid to stream. Downloads deliberately stay on IsAudioAllowed: they fall back to orig_file_cid, which a row missing its track_cid still has, so losing is_streamable must not cost the artist their downloads. Co-Authored-By: Claude Opus 5 --- api/dbv1/tracks.go | 70 ++++++++++++++++++++++--------------- api/v1_track_download.go | 7 ++-- api/v1_track_stream_test.go | 64 +++++++++++++++++++++++++++++++++ 3 files changed, 110 insertions(+), 31 deletions(-) diff --git a/api/dbv1/tracks.go b/api/dbv1/tracks.go index 90eaf33b..2ac9e8ae 100644 --- a/api/dbv1/tracks.go +++ b/api/dbv1/tracks.go @@ -21,15 +21,21 @@ const IncludeID3TagsCtxKey = "includeID3Tags" type Track struct { GetTracksRow - Permalink string `json:"permalink"` - IsStreamable bool `json:"is_streamable"` - Artwork *SquareImage `json:"artwork"` - Stream *MediaLink `json:"stream"` - Download *MediaLink `json:"download"` - Preview *MediaLink `json:"preview"` - UserID trashid.HashId `json:"user_id"` - User User `json:"user"` - Collaborators []User `json:"collaborators"` + Permalink string `json:"permalink"` + IsStreamable bool `json:"is_streamable"` + // IsAudioAllowed reports whether the track and its owner are in good + // standing - not deleted, not deactivated, not delisted. It is the gate on + // serving the artist's audio in any form. IsStreamable narrows it further + // by also requiring a cid to stream, so downloads (which fall back to + // orig_file_cid) must check this rather than IsStreamable. + IsAudioAllowed bool `json:"-"` + Artwork *SquareImage `json:"artwork"` + Stream *MediaLink `json:"stream"` + Download *MediaLink `json:"download"` + Preview *MediaLink `json:"preview"` + UserID trashid.HashId `json:"user_id"` + User User `json:"user"` + Collaborators []User `json:"collaborators"` // PendingCollaborators is populated only on the requester's own tracks (so // the owner's edit form can preserve still-pending invites); empty otherwise. PendingCollaborators []User `json:"pending_collaborators"` @@ -196,26 +202,31 @@ func (q *Queries) TracksKeyed(ctx context.Context, arg TracksParams) (map[int32] } } - // A track is streamable unless it was deleted or its owner is no longer - // active - either the artist deactivated their own account or the - // account was delisted by the trusted notifier. - isStreamable := !rawTrack.IsDelete && !user.IsDeactivated - - // Two reasons to leave a media link nil, both with the same effect: the - // URL should never be handed out, and the endpoints report the track as - // unavailable instead. - // - // A track row can have empty cid columns (e.g. an upload-v2 row whose - // track_cid/orig_file_cid backfill never ran), and signing an empty cid - // produces a content-node URL that is guaranteed to 404. + // The artist's audio is off-limits once the track was deleted or its + // owner stopped being active - either the artist deactivated their own + // account or the account was delisted by the trusted notifier. The cid + // in those rows is real, so a signed URL would still work: the stream + // and download endpoints reject these, but that only closes two routes, + // and anyone reading the track response could otherwise fetch the audio + // straight from the content node. Preview is included because a preview + // clip is still the artist's audio. + isAudioAllowed := !rawTrack.IsDelete && !user.IsDeactivated + + // Streaming needs one more thing: a transcoded cid to point at. An + // upload that never got its track_cid written has audio sitting on the + // content node that no reader can address - nothing to sign, nothing to + // play, and signing an empty cid just produces a URL guaranteed to 404. + // Such rows used to report is_streamable=true, so every client treated + // them as healthy: the player spun on a dead URL, and mobile's + // share-to-story fed that URL to ffmpeg and failed with a generic + // "something went wrong". Say plainly that there is nothing to stream. // - // A non-streamable track is worse: the cid is real, so the signed URL - // works. The stream and download endpoints reject these, but that only - // closes those two routes - anyone reading the track response could - // still fetch the audio straight from the content node. Preview is - // included because a preview clip is still the artist's audio. + // Downloads are deliberately not gated on this - they fall back to + // orig_file_cid, which a row missing its track_cid still has. + isStreamable := isAudioAllowed && rawTrack.TrackCid.String != "" + var stream *MediaLink - if isStreamable && access.Stream && rawTrack.TrackCid.String != "" { + if isStreamable && access.Stream { stream, err = mediaLink(rawTrack.TrackCid.String, rawTrack.TrackID, arg.MyID.(int32), id3Tags) if err != nil { return nil, err @@ -223,7 +234,7 @@ func (q *Queries) TracksKeyed(ctx context.Context, arg TracksParams) (map[int32] } var download *MediaLink - if isStreamable && rawTrack.IsDownloadable && access.Download { + if isAudioAllowed && rawTrack.IsDownloadable && access.Download { cid := rawTrack.OrigFileCid.String if cid == "" { cid = rawTrack.TrackCid.String @@ -237,7 +248,7 @@ func (q *Queries) TracksKeyed(ctx context.Context, arg TracksParams) (map[int32] } var preview *MediaLink - if isStreamable && rawTrack.PreviewCid.String != "" { + if isAudioAllowed && rawTrack.PreviewCid.String != "" { preview, err = mediaLink(rawTrack.PreviewCid.String, rawTrack.TrackID, arg.MyID.(int32), id3Tags) if err != nil { return nil, err @@ -247,6 +258,7 @@ func (q *Queries) TracksKeyed(ctx context.Context, arg TracksParams) (map[int32] track := Track{ GetTracksRow: rawTrack, IsStreamable: isStreamable, + IsAudioAllowed: isAudioAllowed, Permalink: fmt.Sprintf("/%s/%s", user.Handle.String, rawTrack.Slug.String), Artwork: squareImageStruct(rawTrack.CoverArtSizes, rawTrack.CoverArt), Stream: stream, diff --git a/api/v1_track_download.go b/api/v1_track_download.go index 674b7e77..a18d4bdc 100644 --- a/api/v1_track_download.go +++ b/api/v1_track_download.go @@ -44,8 +44,11 @@ func (app *ApiServer) v1TrackDownload(c *fiber.Ctx) error { track := tracks[0] // Same guard as the stream endpoint: a deleted track, or one whose owner is - // no longer active, must not have its audio served here either. - if !track.IsStreamable { + // no longer active, must not have its audio served here either. This checks + // IsAudioAllowed rather than IsStreamable because a download falls back to + // orig_file_cid - a row that never got its track_cid is unstreamable but + // still perfectly downloadable. + if !track.IsAudioAllowed { return fiber.NewError(fiber.StatusNotFound, "track not found") } diff --git a/api/v1_track_stream_test.go b/api/v1_track_stream_test.go index 92d35cb4..71c10012 100644 --- a/api/v1_track_stream_test.go +++ b/api/v1_track_stream_test.go @@ -119,3 +119,67 @@ func TestGetTrackStream_DeletedTrack(t *testing.T) { assert.Equal(t, 404, res.StatusCode) assert.Empty(t, res.Header.Get("Location")) } + +// The track response must say plainly that a row with no track_cid has nothing +// to stream. It used to report is_streamable=true with a null stream link, +// which left every client believing the track was healthy: the player spun on a +// dead URL, and mobile's share-to-story handed that URL to ffmpeg and failed +// with a generic error rather than explaining the track had no audio. +func TestGetTrack_NoCidIsNotStreamable(t *testing.T) { + app := emptyTestApp(t) + fixtures := database.FixtureMap{ + "tracks": []map[string]any{ + { + "track_id": 1, + "owner_id": 1, + "title": "No Cid", + "orig_file_cid": "QmNoCidOriginal", + "is_downloadable": true, + }, + }, + "users": []map[string]any{ + { + "user_id": 1, + "handle": "testuser1", + }, + }, + } + database.Seed(app.pool.Replicas[0], fixtures) + + status, body := testGet(t, app, "/v1/tracks/"+trashid.MustEncodeHashID(1)) + assert.Equal(t, 200, status) + jsonAssert(t, body, map[string]any{ + "data.is_streamable": false, + "data.stream": nil, + }) +} + +// Losing is_streamable must not cost the artist their downloads: a download +// falls back to orig_file_cid, which a row missing its track_cid still has. +func TestGetTrackDownload_NoTrackCidStillDownloadable(t *testing.T) { + app := emptyTestApp(t) + fixtures := database.FixtureMap{ + "tracks": []map[string]any{ + { + "track_id": 1, + "owner_id": 1, + "title": "No Track Cid", + "orig_file_cid": "QmNoCidOriginal", + "orig_filename": "NoCid.wav", + "is_downloadable": true, + }, + }, + "users": []map[string]any{ + { + "user_id": 1, + "handle": "testuser1", + }, + }, + } + database.Seed(app.pool.Replicas[0], fixtures) + req := httptest.NewRequest("GET", "/v1/tracks/"+trashid.MustEncodeHashID(1)+"/download", nil) + res, err := app.Test(req, -1) + assert.NoError(t, err) + assert.Equal(t, 302, res.StatusCode) + assert.Contains(t, res.Header.Get("Location"), "QmNoCidOriginal") +} From 46ea98c621a6353aaaddf9d33ad5e1a66d56222e Mon Sep 17 00:00:00 2001 From: Dylan Jeffers Date: Thu, 3 Sep 2026 14:45:29 -0700 Subject: [PATCH 2/2] feat(api): repair job for tracks indexed without their track_cid track_cid is taken verbatim from the uploader's metadata at index time. The client is supposed to poll the content node until the transcode finishes and then include the resulting cid when it writes the track; when that handshake falls through, the track is indexed with a NULL track_cid. The audio is fine and sitting on the content node - there is just no cid on the row pointing at it. The track looks normal, collects favorites and reposts, and never records a single play. The content node still holds the upload record keyed by the track's audio_upload_id, so reconcile from there, the same way RepairAudioAnalysesJob recovers bpm / musical_key. Unlike that job, this one requires two distinct nodes to agree on the cid before writing it. bpm fills in a display field; track_cid decides which bytes every listener receives for the track. Upload records are replicated across mirrors, so agreement is cheap, and it means a single misbehaving or stale node cannot repoint a track's audio on its own. Tracks that cannot reach quorum, or whose nodes disagree, are logged and left for the next pass rather than repaired on one node's word. Co-Authored-By: Claude Opus 5 --- indexer/indexer.go | 7 + jobs/repair_track_cids.go | 321 +++++++++++++++++++++++++++++++++ jobs/repair_track_cids_test.go | 256 ++++++++++++++++++++++++++ 3 files changed, 584 insertions(+) create mode 100644 jobs/repair_track_cids.go create mode 100644 jobs/repair_track_cids_test.go diff --git a/indexer/indexer.go b/indexer/indexer.go index cca9f8d7..d0a76b3e 100644 --- a/indexer/indexer.go +++ b/indexer/indexer.go @@ -224,6 +224,13 @@ func (ci *CoreIndexer) startParityJobs(ctx context.Context) { // schedule ran every 3 minutes. Needs the SDK for content-node discovery. jobs.NewRepairAudioAnalysesJob(ci.Config, ci.pool, ci.openAudioSDK). ScheduleEvery(ctx, 3*time.Minute) + + // Backfill track_cid for uploads that transcoded successfully but were + // indexed without their cid, leaving the track unplayable. Same + // content-node source as the analysis repair above. Runs less often + // because these are rare and each pass costs a lookup per candidate. + jobs.NewRepairTrackCidsJob(ci.Config, ci.pool, ci.openAudioSDK). + ScheduleEvery(ctx, 15*time.Minute) } func (ci *CoreIndexer) Close() { diff --git a/jobs/repair_track_cids.go b/jobs/repair_track_cids.go new file mode 100644 index 00000000..008c24d3 --- /dev/null +++ b/jobs/repair_track_cids.go @@ -0,0 +1,321 @@ +package jobs + +import ( + "context" + "encoding/json" + "fmt" + "io" + "math/rand" + "net/http" + "strings" + "sync" + "time" + + "api.audius.co/config" + "api.audius.co/database" + "api.audius.co/logging" + connect "connectrpc.com/connect" + ethv1 "github.com/OpenAudio/go-openaudio/pkg/api/eth/v1" + "github.com/OpenAudio/go-openaudio/pkg/sdk" + "go.uber.org/zap" +) + +// RepairTrackCidsJob backfills tracks whose audio finished transcoding but +// whose track_cid never made it onto the track entity. +// +// track_cid is taken verbatim from the uploader's metadata at index time. The +// client is supposed to poll the content node until the transcode finishes and +// then include the resulting cid when it writes the track, but when that +// handshake falls through the track is indexed with a NULL track_cid. The audio +// is fine and sitting on the content node - there is simply no cid on the row +// pointing at it, so nothing can be signed and nothing can be played. The +// track is silently dead: it looks normal, collects favorites and reposts, and +// never accumulates a single play. +// +// The content node is the authoritative source for what it produced, and it +// still holds the upload record keyed by the track's audio_upload_id. This job +// reconciles the gap from there, the same way RepairAudioAnalysesJob recovers +// bpm / musical_key. +// +// Each pass: +// 1. Selects up to trackCidBatchSize current, undeleted tracks with a NULL +// track_cid and an audio_upload_id to look up (newest first). +// 2. Picks up to trackCidMaxNodes random registered content nodes. +// 3. Queries nodes for each upload record until trackCidQuorum of them agree +// on the same transcoded cid. +// 4. Writes track_cid, committing per track. +type RepairTrackCidsJob struct { + pool database.DbPool + logger *zap.Logger + sdk *sdk.OpenAudioSDK + httpClient *http.Client + + mutex sync.Mutex + isRunning bool +} + +const ( + // trackCidBatchSize matches RepairAudioAnalysesJob's batch. + trackCidBatchSize = 1000 + // trackCidMaxNodes bounds how many nodes a single pass will ask. + trackCidMaxNodes = 5 + // trackCidNodeTimeout matches RepairAudioAnalysesJob's per-request budget. + trackCidNodeTimeout = 5 * time.Second + // trackCidQuorum is how many distinct content nodes must report the same + // cid before it is written. + // + // This is a higher bar than the bpm / musical_key repair asks for, and + // deliberately so: those fill in a display field, while track_cid decides + // which bytes every listener receives for this track. Upload records are + // replicated across mirrors, so agreement is cheap to obtain and means a + // single misbehaving or out-of-date node cannot repoint a track's audio on + // its own. A track that cannot reach quorum is left alone for the next pass + // rather than repaired from one node's word. + trackCidQuorum = 2 +) + +func NewRepairTrackCidsJob(cfg config.Config, pool database.DbPool, oaSDK *sdk.OpenAudioSDK) *RepairTrackCidsJob { + return &RepairTrackCidsJob{ + pool: pool, + logger: logging.NewZapLogger(cfg).Named("RepairTrackCidsJob"), + sdk: oaSDK, + httpClient: &http.Client{Timeout: trackCidNodeTimeout}, + } +} + +// ScheduleEvery runs the job every `interval` until the context is cancelled. +func (j *RepairTrackCidsJob) ScheduleEvery(ctx context.Context, interval time.Duration) *RepairTrackCidsJob { + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ticker.C: + j.Run(ctx) + case <-ctx.Done(): + j.logger.Info("Job shutting down") + return + } + } + }() + return j +} + +// Run executes the job once. +func (j *RepairTrackCidsJob) Run(ctx context.Context) { + if err := j.run(ctx); err != nil { + j.logger.Error("Job run failed", zap.Error(err)) + } +} + +func (j *RepairTrackCidsJob) run(ctx context.Context) error { + j.mutex.Lock() + if j.isRunning { + j.mutex.Unlock() + return fmt.Errorf("job is already running") + } + j.isRunning = true + j.mutex.Unlock() + defer func() { + j.mutex.Lock() + j.isRunning = false + j.mutex.Unlock() + }() + + tracks, err := j.queryTracks(ctx) + if err != nil { + return fmt.Errorf("query tracks: %w", err) + } + if len(tracks) == 0 { + return nil + } + + nodes, err := j.selectContentNodes(ctx) + if err != nil { + return fmt.Errorf("select content nodes: %w", err) + } + if len(nodes) < trackCidQuorum { + j.logger.Warn("not enough content nodes to reach quorum; skipping pass", + zap.Int("nodes", len(nodes)), zap.Int("quorum", trackCidQuorum)) + return nil + } + + repaired := 0 + for _, t := range tracks { + ok, err := j.repairTrackCid(ctx, t, nodes) + if err != nil { + j.logger.Error("repairing track cid failed", + zap.Int64("track_id", t.TrackID), zap.Error(err)) + continue + } + if ok { + repaired++ + } + } + + j.logger.Info("Repaired track cids", + zap.Int("candidates", len(tracks)), + zap.Int("repaired", repaired)) + return nil +} + +type cidlessTrack struct { + TrackID int64 + AudioUploadID string +} + +// queryTracks selects tracks that have no playable audio pointer but do carry +// the upload id needed to find one. Deleted tracks are skipped - their audio is +// meant to be unreachable - as are stems and rows with no audio_upload_id, +// which are legacy uploads with nothing to look up. +func (j *RepairTrackCidsJob) queryTracks(ctx context.Context) ([]cidlessTrack, error) { + rows, err := j.pool.Query(ctx, ` + SELECT track_id, audio_upload_id + FROM tracks + WHERE is_current = true + AND is_delete = false + AND track_cid IS NULL + AND audio_upload_id IS NOT NULL + AND audio_upload_id <> '' + ORDER BY created_at DESC + LIMIT $1 + `, trackCidBatchSize) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []cidlessTrack + for rows.Next() { + var t cidlessTrack + if err := rows.Scan(&t.TrackID, &t.AudioUploadID); err != nil { + return nil, err + } + out = append(out, t) + } + return out, rows.Err() +} + +// selectContentNodes takes up to trackCidMaxNodes random registered +// content-node endpoints. Mirrors RepairAudioAnalysesJob.selectContentNodes. +func (j *RepairTrackCidsJob) selectContentNodes(ctx context.Context) ([]string, error) { + resp, err := j.sdk.Eth.GetRegisteredEndpoints(ctx, connect.NewRequest(ðv1.GetRegisteredEndpointsRequest{})) + if err != nil { + return nil, err + } + if resp == nil || resp.Msg == nil { + return nil, fmt.Errorf("GetRegisteredEndpoints returned nil response") + } + + var endpoints []string + for _, node := range resp.Msg.Endpoints { + if node.ServiceType != "content-node" { + continue + } + ep := strings.TrimRight(strings.ToLower(strings.TrimSpace(node.Endpoint)), "/") + if ep != "" { + endpoints = append(endpoints, ep) + } + } + + rand.Shuffle(len(endpoints), func(a, b int) { + endpoints[a], endpoints[b] = endpoints[b], endpoints[a] + }) + if len(endpoints) > trackCidMaxNodes { + endpoints = endpoints[:trackCidMaxNodes] + } + return endpoints, nil +} + +// uploadRecord is the subset of mediorum's /uploads/:id payload this job needs. +type uploadRecord struct { + Status string `json:"status"` + TranscodeResults map[string]string `json:"results"` +} + +// transcodedCid returns the 320kbps cid of a finished transcode, or "" when the +// upload has not produced one yet. +func (u uploadRecord) transcodedCid() string { + if u.Status != "done" { + return "" + } + return strings.TrimSpace(u.TranscodeResults["320"]) +} + +// repairTrackCid asks content nodes for one track's upload record and writes +// the transcoded cid once trackCidQuorum nodes agree on it. Returns true when +// the track was repaired. +func (j *RepairTrackCidsJob) repairTrackCid(ctx context.Context, t cidlessTrack, nodes []string) (bool, error) { + votes := make(map[string]int, 2) + for _, node := range nodes { + cid, ok := j.fetchTranscodedCid(ctx, node, t.AudioUploadID) + if !ok || cid == "" { + // Transport error, no record, or a transcode that has not finished: + // nothing to count. Ask the next node. + continue + } + + votes[cid]++ + if votes[cid] < trackCidQuorum { + continue + } + + if err := j.applyTrackCid(ctx, t.TrackID, cid); err != nil { + return false, fmt.Errorf("update track %d: %w", t.TrackID, err) + } + j.logger.Info("repaired track cid", + zap.Int64("track_id", t.TrackID), + zap.String("audio_upload_id", t.AudioUploadID), + zap.String("track_cid", cid)) + return true, nil + } + + if len(votes) > 1 { + // Nodes disagreed about what this upload transcoded to. Never guess + // which one is right; leave the row alone and say so loudly. + j.logger.Warn("content nodes disagree on transcoded cid; leaving track unrepaired", + zap.Int64("track_id", t.TrackID), + zap.String("audio_upload_id", t.AudioUploadID), + zap.Any("votes", votes)) + } + return false, nil +} + +// fetchTranscodedCid GETs one node's upload record. ok=false signals a +// transport/non-2xx error; ok=true with an empty cid means the node answered +// but the upload has not finished transcoding. +func (j *RepairTrackCidsJob) fetchTranscodedCid(ctx context.Context, node, uploadID string) (cid string, ok bool) { + endpoint := fmt.Sprintf("%s/uploads/%s", node, uploadID) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) + if err != nil { + return "", false + } + resp, err := j.httpClient.Do(req) + if err != nil { + return "", false + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + io.Copy(io.Discard, io.LimitReader(resp.Body, 512)) + return "", false + } + body, err := io.ReadAll(resp.Body) + if err != nil { + return "", false + } + var parsed uploadRecord + if err := json.Unmarshal(body, &parsed); err != nil { + return "", false + } + return parsed.transcodedCid(), true +} + +// applyTrackCid writes the cid, re-checking that the row is still cidless so a +// concurrent indexer write of the real metadata always wins. +func (j *RepairTrackCidsJob) applyTrackCid(ctx context.Context, trackID int64, cid string) error { + _, err := j.pool.Exec(ctx, ` + UPDATE tracks SET track_cid = $2 + WHERE track_id = $1 AND is_current = true AND track_cid IS NULL + `, trackID, cid) + return err +} diff --git a/jobs/repair_track_cids_test.go b/jobs/repair_track_cids_test.go new file mode 100644 index 00000000..1d628eab --- /dev/null +++ b/jobs/repair_track_cids_test.go @@ -0,0 +1,256 @@ +package jobs + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" + + "api.audius.co/database" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/zap" +) + +func TestTranscodedCid(t *testing.T) { + // A finished transcode hands back its 320kbps cid. + done := uploadRecord{Status: "done", TranscodeResults: map[string]string{"320": "QmDone"}} + assert.Equal(t, "QmDone", done.transcodedCid()) + + // An upload still in flight has nothing to offer, even if a partial result + // is already present - writing that would point the track at a half-built + // blob. + busy := uploadRecord{Status: "busy", TranscodeResults: map[string]string{"320": "QmPartial"}} + assert.Equal(t, "", busy.transcodedCid()) + + // A finished upload with no 320 result is not a repair candidate. + empty := uploadRecord{Status: "done", TranscodeResults: map[string]string{}} + assert.Equal(t, "", empty.transcodedCid()) + + whitespace := uploadRecord{Status: "done", TranscodeResults: map[string]string{"320": " QmPadded "}} + assert.Equal(t, "QmPadded", whitespace.transcodedCid()) +} + +func newTrackCidJob(pool database.DbPool) *RepairTrackCidsJob { + return &RepairTrackCidsJob{ + pool: pool, + logger: zap.NewNop(), + httpClient: &http.Client{Timeout: 2 * time.Second}, + } +} + +func TestFetchTranscodedCid(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + assert.Equal(t, "/uploads/up-1", r.URL.Path) + w.Write([]byte(`{"status":"done","results":{"320":"QmGood"}}`)) + })) + defer srv.Close() + + cid, ok := newTrackCidJob(nil).fetchTranscodedCid(context.Background(), srv.URL, "up-1") + require.True(t, ok) + assert.Equal(t, "QmGood", cid) +} + +func TestFetchTranscodedCidNon2xxIsNotOk(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "not found", http.StatusNotFound) + })) + defer srv.Close() + + _, ok := newTrackCidJob(nil).fetchTranscodedCid(context.Background(), srv.URL, "missing") + assert.False(t, ok, "a node that does not hold the upload must not count as a vote") +} + +func TestFetchTranscodedCidUnfinishedTranscode(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte(`{"status":"busy","results":{}}`)) + })) + defer srv.Close() + + cid, ok := newTrackCidJob(nil).fetchTranscodedCid(context.Background(), srv.URL, "up-1") + assert.True(t, ok, "the node answered") + assert.Equal(t, "", cid, "but has no cid to offer yet") +} + +// nodeServing returns a content node stub that reports the given upload record +// JSON, and counts how many times it was asked. +func nodeServing(t *testing.T, body string, calls *int) string { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + *calls++ + w.Write([]byte(body)) + })) + t.Cleanup(srv.Close) + return srv.URL +} + +func seedCidlessTrack(t *testing.T, pool *pgxpool.Pool, uploadID string) { + t.Helper() + database.Seed(pool, database.FixtureMap{ + "users": { + {"user_id": 1, "wallet": "0x01", "handle": "testuser1"}, + }, + "tracks": { + {"track_id": 100, "owner_id": 1, "title": "No Cid", "audio_upload_id": uploadID}, + }, + }) +} + +func trackCidOf(t *testing.T, pool *pgxpool.Pool, trackID int64) *string { + t.Helper() + var cid *string + require.NoError(t, pool.QueryRow(context.Background(), + "SELECT track_cid FROM tracks WHERE track_id = $1", trackID).Scan(&cid)) + return cid +} + +// The whole point of the job: an upload that transcoded fine but was indexed +// without its cid gets its audio pointer back. +func TestRepairTrackCidsJob_RepairsOnQuorum(t *testing.T) { + pool := database.CreateTestDatabase(t, "test_jobs") + defer pool.Close() + seedCidlessTrack(t, pool, "up-1") + + body := `{"status":"done","results":{"320":"QmAgreed"}}` + calls := 0 + nodes := []string{ + nodeServing(t, body, &calls), + nodeServing(t, body, &calls), + nodeServing(t, body, &calls), + } + + job := newTrackCidJob(pool) + repaired, err := job.repairTrackCid(context.Background(), + cidlessTrack{TrackID: 100, AudioUploadID: "up-1"}, nodes) + require.NoError(t, err) + assert.True(t, repaired) + + cid := trackCidOf(t, pool, 100) + require.NotNil(t, cid) + assert.Equal(t, "QmAgreed", *cid) + assert.Equal(t, trackCidQuorum, calls, "stop asking nodes once quorum is reached") +} + +// track_cid decides which bytes every listener receives, so one node's word is +// never enough to set it. +func TestRepairTrackCidsJob_SingleNodeCannotRepair(t *testing.T) { + pool := database.CreateTestDatabase(t, "test_jobs") + defer pool.Close() + seedCidlessTrack(t, pool, "up-1") + + calls := 0 + nodes := []string{nodeServing(t, `{"status":"done","results":{"320":"QmLonely"}}`, &calls)} + + job := newTrackCidJob(pool) + repaired, err := job.repairTrackCid(context.Background(), + cidlessTrack{TrackID: 100, AudioUploadID: "up-1"}, nodes) + require.NoError(t, err) + assert.False(t, repaired) + assert.Nil(t, trackCidOf(t, pool, 100)) +} + +// Disagreement means something is wrong upstream. Leave the row alone rather +// than picking a winner. +func TestRepairTrackCidsJob_DisagreementLeavesTrackAlone(t *testing.T) { + pool := database.CreateTestDatabase(t, "test_jobs") + defer pool.Close() + seedCidlessTrack(t, pool, "up-1") + + calls := 0 + nodes := []string{ + nodeServing(t, `{"status":"done","results":{"320":"QmOne"}}`, &calls), + nodeServing(t, `{"status":"done","results":{"320":"QmTwo"}}`, &calls), + nodeServing(t, `{"status":"done","results":{"320":"QmThree"}}`, &calls), + } + + job := newTrackCidJob(pool) + repaired, err := job.repairTrackCid(context.Background(), + cidlessTrack{TrackID: 100, AudioUploadID: "up-1"}, nodes) + require.NoError(t, err) + assert.False(t, repaired) + assert.Nil(t, trackCidOf(t, pool, 100)) +} + +// Unreachable nodes must not block a repair that the reachable ones agree on. +func TestRepairTrackCidsJob_SkipsUnreachableNodes(t *testing.T) { + pool := database.CreateTestDatabase(t, "test_jobs") + defer pool.Close() + seedCidlessTrack(t, pool, "up-1") + + calls := 0 + dead := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "boom", http.StatusInternalServerError) + })) + defer dead.Close() + + body := `{"status":"done","results":{"320":"QmAgreed"}}` + nodes := []string{ + dead.URL, + nodeServing(t, body, &calls), + nodeServing(t, body, &calls), + } + + job := newTrackCidJob(pool) + repaired, err := job.repairTrackCid(context.Background(), + cidlessTrack{TrackID: 100, AudioUploadID: "up-1"}, nodes) + require.NoError(t, err) + assert.True(t, repaired) + + cid := trackCidOf(t, pool, 100) + require.NotNil(t, cid) + assert.Equal(t, "QmAgreed", *cid) +} + +// A track that already has audio is not a candidate, and neither is a deleted +// one - its audio is meant to stay unreachable. +func TestRepairTrackCidsJob_QueryTracksSelectsOnlyRepairable(t *testing.T) { + pool := database.CreateTestDatabase(t, "test_jobs") + defer pool.Close() + + database.Seed(pool, database.FixtureMap{ + "users": { + {"user_id": 1, "wallet": "0x01", "handle": "testuser1"}, + }, + "tracks": { + {"track_id": 100, "owner_id": 1, "title": "Repairable", "audio_upload_id": "up-1"}, + {"track_id": 101, "owner_id": 1, "title": "Has Cid", "audio_upload_id": "up-2", "track_cid": "QmAlready"}, + {"track_id": 102, "owner_id": 1, "title": "Deleted", "audio_upload_id": "up-3", "is_delete": true}, + {"track_id": 103, "owner_id": 1, "title": "No Upload Id"}, + }, + }) + + job := newTrackCidJob(pool) + tracks, err := job.queryTracks(context.Background()) + require.NoError(t, err) + + ids := []int64{} + for _, tr := range tracks { + ids = append(ids, tr.TrackID) + } + assert.Equal(t, []int64{100}, ids, fmt.Sprintf("got %+v", tracks)) +} + +// The indexer writing real metadata must always beat the repair job. +func TestApplyTrackCidDoesNotOverwriteExistingCid(t *testing.T) { + pool := database.CreateTestDatabase(t, "test_jobs") + defer pool.Close() + + database.Seed(pool, database.FixtureMap{ + "users": { + {"user_id": 1, "wallet": "0x01", "handle": "testuser1"}, + }, + "tracks": { + {"track_id": 100, "owner_id": 1, "title": "Has Cid", "track_cid": "QmReal"}, + }, + }) + + job := newTrackCidJob(pool) + require.NoError(t, job.applyTrackCid(context.Background(), 100, "QmRepaired")) + + cid := trackCidOf(t, pool, 100) + require.NotNil(t, cid) + assert.Equal(t, "QmReal", *cid) +}