From b32bd7af85e97e177ac8367b68583cd836ba9873 Mon Sep 17 00:00:00 2001 From: Dylan Jeffers Date: Thu, 3 Sep 2026 14:41:01 -0700 Subject: [PATCH 1/5] 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/5] 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) +} From c747c30eb27a9b7147d3b932946902f9df4d8805 Mon Sep 17 00:00:00 2001 From: Dylan Jeffers Date: Wed, 23 Sep 2026 16:52:05 -0700 Subject: [PATCH 3/5] test(api): owner can download a track with no track_cid Co-Authored-By: Claude Opus 5.5 --- api/v1_track_download_test.go | 29 +++++++++++++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/api/v1_track_download_test.go b/api/v1_track_download_test.go index b7b3daec..f6071402 100644 --- a/api/v1_track_download_test.go +++ b/api/v1_track_download_test.go @@ -198,3 +198,32 @@ func TestGetTrackDownload_FilenameFallsBackToMp3(t *testing.T) { assert.Contains(t, location, "tracks/cidstream/QmTranscode") assert.Contains(t, location, "filename=Vol.+2.mp3") } + +// The owner can download a track with no track_cid and downloads off; the link +// falls back to orig_file_cid. +func TestGetTrackDownload_OwnerOfCidlessTrack(t *testing.T) { + app := emptyTestApp(t) + database.Seed(app.pool.Replicas[0], database.FixtureMap{ + "tracks": []map[string]any{ + { + "track_id": 1, + "owner_id": 1, + "title": "No Track Cid", + "orig_file_cid": "QmOriginal", + "orig_filename": "NoCid.wav", + "is_downloadable": false, + }, + }, + "users": []map[string]any{ + {"user_id": 1, "handle": "artist", "wallet": ownerWallet}, + }, + }) + path := "/v1/tracks/" + trashid.MustEncodeHashID(1) + "/download" + + status, location := downloadWithWallet(t, app, path, ownerWallet) + assert.Equal(t, 302, status) + assert.Contains(t, location, "tracks/cidstream/QmOriginal") + + status, _ = downloadWithWallet(t, app, path, "") + assert.Equal(t, 404, status, "anonymous") +} From 01c67eae2142491dc66b0c3e47adc5e19cc1047c Mon Sep 17 00:00:00 2001 From: Dylan Jeffers Date: Wed, 23 Sep 2026 16:52:42 -0700 Subject: [PATCH 4/5] fix(jobs): back off unrepairable tracks in the track_cid repair job Tracks that can't be repaired (transcode never finished, no quorum) were selected again every pass, so enough of them would fill the newest-1000 batch and block older repairable rows while re-querying nodes every 15 minutes. Failed tracks now back off from 1h, doubling to a 24h cap, and are excluded from the batch query while backed off. Also exclude stems from the candidate query, as the doc comment said. Co-Authored-By: Claude Opus 5.5 --- jobs/repair_track_cids.go | 64 +++++++++++++++++++++++++++++----- jobs/repair_track_cids_test.go | 23 +++++++++++- 2 files changed, 78 insertions(+), 9 deletions(-) diff --git a/jobs/repair_track_cids.go b/jobs/repair_track_cids.go index 008c24d3..338718fd 100644 --- a/jobs/repair_track_cids.go +++ b/jobs/repair_track_cids.go @@ -52,6 +52,16 @@ type RepairTrackCidsJob struct { mutex sync.Mutex isRunning bool + + // retries holds tracks a pass could not repair (transcode not finished, + // no quorum, or an error). They are skipped until their backoff expires so + // they don't fill every batch. In memory only: a restart retries them once. + retries map[int64]trackCidRetry +} + +type trackCidRetry struct { + failures int + next time.Time } const ( @@ -72,6 +82,11 @@ const ( // 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 + + // Backoff for tracks a pass could not repair: doubles from + // trackCidRetryBase per failure, capped at trackCidRetryMax. + trackCidRetryBase = time.Hour + trackCidRetryMax = 24 * time.Hour ) func NewRepairTrackCidsJob(cfg config.Config, pool database.DbPool, oaSDK *sdk.OpenAudioSDK) *RepairTrackCidsJob { @@ -122,7 +137,7 @@ func (j *RepairTrackCidsJob) run(ctx context.Context) error { j.mutex.Unlock() }() - tracks, err := j.queryTracks(ctx) + tracks, err := j.queryTracks(ctx, j.backedOffTrackIDs(time.Now())) if err != nil { return fmt.Errorf("query tracks: %w", err) } @@ -146,10 +161,12 @@ func (j *RepairTrackCidsJob) run(ctx context.Context) error { if err != nil { j.logger.Error("repairing track cid failed", zap.Int64("track_id", t.TrackID), zap.Error(err)) - continue } if ok { repaired++ + delete(j.retries, t.TrackID) + } else { + j.recordFailure(t.TrackID, time.Now()) } } @@ -164,22 +181,26 @@ type cidlessTrack struct { 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) { +// queryTracks selects tracks with no track_cid but an audio_upload_id to look +// up. Skips deleted tracks (their audio must stay unreachable), stems, rows +// with no audio_upload_id (legacy uploads), and the ids in skip. +func (j *RepairTrackCidsJob) queryTracks(ctx context.Context, skip []int64) ([]cidlessTrack, error) { + if skip == nil { + skip = []int64{} + } rows, err := j.pool.Query(ctx, ` SELECT track_id, audio_upload_id FROM tracks WHERE is_current = true AND is_delete = false + AND stem_of IS NULL AND track_cid IS NULL AND audio_upload_id IS NOT NULL AND audio_upload_id <> '' + AND NOT (track_id = ANY($2::bigint[])) ORDER BY created_at DESC LIMIT $1 - `, trackCidBatchSize) + `, trackCidBatchSize, skip) if err != nil { return nil, err } @@ -196,6 +217,33 @@ func (j *RepairTrackCidsJob) queryTracks(ctx context.Context) ([]cidlessTrack, e return out, rows.Err() } +// backedOffTrackIDs returns the tracks still inside their retry backoff at now. +func (j *RepairTrackCidsJob) backedOffTrackIDs(now time.Time) []int64 { + ids := make([]int64, 0, len(j.retries)) + for id, r := range j.retries { + if now.Before(r.next) { + ids = append(ids, id) + } + } + return ids +} + +// recordFailure pushes a track's next attempt out by the backoff for its +// failure count. +func (j *RepairTrackCidsJob) recordFailure(trackID int64, now time.Time) { + if j.retries == nil { + j.retries = map[int64]trackCidRetry{} + } + r := j.retries[trackID] + r.failures++ + delay := trackCidRetryMax + if r.failures <= 5 { + delay = min(trackCidRetryBase<<(r.failures-1), trackCidRetryMax) + } + r.next = now.Add(delay) + j.retries[trackID] = r +} + // selectContentNodes takes up to trackCidMaxNodes random registered // content-node endpoints. Mirrors RepairAudioAnalysesJob.selectContentNodes. func (j *RepairTrackCidsJob) selectContentNodes(ctx context.Context) ([]string, error) { diff --git a/jobs/repair_track_cids_test.go b/jobs/repair_track_cids_test.go index 1d628eab..7d71f8cf 100644 --- a/jobs/repair_track_cids_test.go +++ b/jobs/repair_track_cids_test.go @@ -219,11 +219,13 @@ func TestRepairTrackCidsJob_QueryTracksSelectsOnlyRepairable(t *testing.T) { {"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"}, + {"track_id": 104, "owner_id": 1, "title": "Stem", "audio_upload_id": "up-4", "stem_of": `{"parent_track_id": 100, "category": "other"}`}, + {"track_id": 105, "owner_id": 1, "title": "Backed Off", "audio_upload_id": "up-5"}, }, }) job := newTrackCidJob(pool) - tracks, err := job.queryTracks(context.Background()) + tracks, err := job.queryTracks(context.Background(), []int64{105}) require.NoError(t, err) ids := []int64{} @@ -233,6 +235,25 @@ func TestRepairTrackCidsJob_QueryTracksSelectsOnlyRepairable(t *testing.T) { assert.Equal(t, []int64{100}, ids, fmt.Sprintf("got %+v", tracks)) } +// Tracks that could not be repaired are skipped with a growing backoff. +func TestRepairTrackCidsJob_BackoffAfterFailure(t *testing.T) { + job := newTrackCidJob(nil) + now := time.Now() + + job.recordFailure(100, now) + assert.Equal(t, []int64{100}, job.backedOffTrackIDs(now)) + assert.Empty(t, job.backedOffTrackIDs(now.Add(trackCidRetryBase))) + + job.recordFailure(100, now) + assert.Equal(t, []int64{100}, job.backedOffTrackIDs(now.Add(trackCidRetryBase))) + assert.Empty(t, job.backedOffTrackIDs(now.Add(2*trackCidRetryBase))) + + for i := 0; i < 10; i++ { + job.recordFailure(100, now) + } + assert.Empty(t, job.backedOffTrackIDs(now.Add(trackCidRetryMax)), "backoff is capped") +} + // The indexer writing real metadata must always beat the repair job. func TestApplyTrackCidDoesNotOverwriteExistingCid(t *testing.T) { pool := database.CreateTestDatabase(t, "test_jobs") From 44bde2c18789a3f0d75f8e812f065db72fdd4b6a Mon Sep 17 00:00:00 2001 From: Dylan Jeffers Date: Wed, 23 Sep 2026 16:54:11 -0700 Subject: [PATCH 5/5] chore(api): shorten comments in the no-track_cid change and repair job Co-Authored-By: Claude Opus 5.5 --- api/dbv1/tracks.go | 32 +++++++------------------- api/v1_track_download.go | 8 +++---- api/v1_track_stream_test.go | 9 ++------ indexer/indexer.go | 7 +++--- jobs/repair_track_cids.go | 41 +++++++++++----------------------- jobs/repair_track_cids_test.go | 19 ++++++---------- 6 files changed, 36 insertions(+), 80 deletions(-) diff --git a/api/dbv1/tracks.go b/api/dbv1/tracks.go index fca73fc0..ca45c3c1 100644 --- a/api/dbv1/tracks.go +++ b/api/dbv1/tracks.go @@ -23,11 +23,9 @@ type Track struct { 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 is false when the track is deleted or its owner is + // deactivated or delisted. IsStreamable also requires a track_cid, so + // downloads (which fall back to orig_file_cid) check this instead. IsAudioAllowed bool `json:"-"` Artwork *SquareImage `json:"artwork"` Stream *MediaLink `json:"stream"` @@ -202,27 +200,13 @@ func (q *Queries) TracksKeyed(ctx context.Context, arg TracksParams) (map[int32] } } - // 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. + // No media links (stream, download or preview) for a deleted track or + // an inactive owner. The cid is real, so a signed link in the response + // would bypass the stream and download endpoint checks. 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. - // - // Downloads are deliberately not gated on this - they fall back to - // orig_file_cid, which a row missing its track_cid still has. + // Streaming also needs a track_cid; without one there is nothing to + // play. Downloads don't, since they fall back to orig_file_cid. isStreamable := isAudioAllowed && rawTrack.TrackCid.String != "" var stream *MediaLink diff --git a/api/v1_track_download.go b/api/v1_track_download.go index e022ac8b..bc507f9e 100644 --- a/api/v1_track_download.go +++ b/api/v1_track_download.go @@ -55,11 +55,9 @@ 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. 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. + // Deleted tracks and inactive owners get 404, as on the stream endpoint. + // Checks IsAudioAllowed rather than IsStreamable because downloads fall + // back to orig_file_cid. 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 71c10012..8102e5e0 100644 --- a/api/v1_track_stream_test.go +++ b/api/v1_track_stream_test.go @@ -120,11 +120,7 @@ func TestGetTrackStream_DeletedTrack(t *testing.T) { 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. +// A track with no track_cid reports is_streamable=false and a null stream. func TestGetTrack_NoCidIsNotStreamable(t *testing.T) { app := emptyTestApp(t) fixtures := database.FixtureMap{ @@ -154,8 +150,7 @@ func TestGetTrack_NoCidIsNotStreamable(t *testing.T) { }) } -// 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. +// A track with no track_cid is still downloadable via orig_file_cid. func TestGetTrackDownload_NoTrackCidStillDownloadable(t *testing.T) { app := emptyTestApp(t) fixtures := database.FixtureMap{ diff --git a/indexer/indexer.go b/indexer/indexer.go index 71f64c11..4011169a 100644 --- a/indexer/indexer.go +++ b/indexer/indexer.go @@ -230,10 +230,9 @@ func (ci *CoreIndexer) startParityJobs(ctx context.Context) { 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. + // Backfill track_cid for uploads that transcoded but were indexed without + // it. Runs less often than the analysis repair since these are rare and + // each candidate costs a content-node lookup. jobs.NewRepairTrackCidsJob(ci.Config, ci.pool, ci.openAudioSDK). ScheduleEvery(ctx, 15*time.Minute) } diff --git a/jobs/repair_track_cids.go b/jobs/repair_track_cids.go index 338718fd..e4d40f80 100644 --- a/jobs/repair_track_cids.go +++ b/jobs/repair_track_cids.go @@ -20,22 +20,14 @@ import ( "go.uber.org/zap" ) -// RepairTrackCidsJob backfills tracks whose audio finished transcoding but -// whose track_cid never made it onto the track entity. +// RepairTrackCidsJob backfills track_cid for tracks whose audio finished +// transcoding but were indexed without the 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, 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. +// track_cid comes from the uploader's metadata at index time. When the client +// writes the track without it, the row has nothing to sign and the track is +// unplayable. The content node still has the upload record keyed by +// audio_upload_id, so this job reads the cid from there, like +// RepairAudioAnalysesJob does for bpm / musical_key. // // Each pass: // 1. Selects up to trackCidBatchSize current, undeleted tracks with a NULL @@ -72,15 +64,10 @@ const ( // 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. + // cid before it is written. Stricter than the bpm / musical_key repair + // because track_cid selects the audio every listener gets, so one stale or + // faulty node can't set it. Upload records are replicated, so quorum is + // cheap. Tracks without quorum are retried later. trackCidQuorum = 2 // Backoff for tracks a pass could not repair: doubles from @@ -298,8 +285,7 @@ func (j *RepairTrackCidsJob) repairTrackCid(ctx context.Context, t cidlessTrack, 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. + // Transport error, no record, or transcode not finished. continue } @@ -319,8 +305,7 @@ func (j *RepairTrackCidsJob) repairTrackCid(ctx context.Context, t cidlessTrack, } 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. + // Nodes disagree on the cid. Leave the row for manual review. 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), diff --git a/jobs/repair_track_cids_test.go b/jobs/repair_track_cids_test.go index 7d71f8cf..4bf69ea8 100644 --- a/jobs/repair_track_cids_test.go +++ b/jobs/repair_track_cids_test.go @@ -20,9 +20,7 @@ func TestTranscodedCid(t *testing.T) { 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. + // An unfinished upload yields no cid, even with a partial result present. busy := uploadRecord{Status: "busy", TranscodeResults: map[string]string{"320": "QmPartial"}} assert.Equal(t, "", busy.transcodedCid()) @@ -107,8 +105,7 @@ func trackCidOf(t *testing.T, pool *pgxpool.Pool, trackID int64) *string { return cid } -// The whole point of the job: an upload that transcoded fine but was indexed -// without its cid gets its audio pointer back. +// A cidless track is repaired when nodes reach quorum. func TestRepairTrackCidsJob_RepairsOnQuorum(t *testing.T) { pool := database.CreateTestDatabase(t, "test_jobs") defer pool.Close() @@ -134,8 +131,7 @@ func TestRepairTrackCidsJob_RepairsOnQuorum(t *testing.T) { 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. +// One node is not enough to set track_cid. func TestRepairTrackCidsJob_SingleNodeCannotRepair(t *testing.T) { pool := database.CreateTestDatabase(t, "test_jobs") defer pool.Close() @@ -152,8 +148,7 @@ func TestRepairTrackCidsJob_SingleNodeCannotRepair(t *testing.T) { assert.Nil(t, trackCidOf(t, pool, 100)) } -// Disagreement means something is wrong upstream. Leave the row alone rather -// than picking a winner. +// Disagreeing nodes leave the row unchanged. func TestRepairTrackCidsJob_DisagreementLeavesTrackAlone(t *testing.T) { pool := database.CreateTestDatabase(t, "test_jobs") defer pool.Close() @@ -204,8 +199,8 @@ func TestRepairTrackCidsJob_SkipsUnreachableNodes(t *testing.T) { 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. +// Only current, undeleted, non-stem tracks with no track_cid, an upload id, +// and no active backoff are candidates. func TestRepairTrackCidsJob_QueryTracksSelectsOnlyRepairable(t *testing.T) { pool := database.CreateTestDatabase(t, "test_jobs") defer pool.Close() @@ -254,7 +249,7 @@ func TestRepairTrackCidsJob_BackoffAfterFailure(t *testing.T) { assert.Empty(t, job.backedOffTrackIDs(now.Add(trackCidRetryMax)), "backoff is capped") } -// The indexer writing real metadata must always beat the repair job. +// applyTrackCid never overwrites an existing cid. func TestApplyTrackCidDoesNotOverwriteExistingCid(t *testing.T) { pool := database.CreateTestDatabase(t, "test_jobs") defer pool.Close()