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") +} 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) +}