diff --git a/api/dbv1/tracks.go b/api/dbv1/tracks.go index d9cf1fb1..ca45c3c1 100644 --- a/api/dbv1/tracks.go +++ b/api/dbv1/tracks.go @@ -21,15 +21,19 @@ 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 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"` + 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,17 +200,17 @@ 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 + // 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 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 != "" - // Media links stay nil when there is no cid to sign (the URL would 404) - // or the track is not streamable (the cid is real, so a signed URL would - // bypass the stream and download endpoint checks). Previews count as - // the artist's audio too. 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 @@ -214,7 +218,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 { if cid := rawTrack.DownloadCid(); cid != "" { download, err = mediaLink(cid, rawTrack.TrackID, arg.MyID.(int32), nil) if err != nil { @@ -224,7 +228,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 @@ -234,6 +238,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 9797c5fd..b5f4c35f 100644 --- a/api/v1_track_download.go +++ b/api/v1_track_download.go @@ -53,9 +53,10 @@ 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 { + // 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_download_test.go b/api/v1_track_download_test.go index 0f171db0..d54d3b25 100644 --- a/api/v1_track_download_test.go +++ b/api/v1_track_download_test.go @@ -193,3 +193,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") +} diff --git a/api/v1_track_stream_test.go b/api/v1_track_stream_test.go index 92d35cb4..8102e5e0 100644 --- a/api/v1_track_stream_test.go +++ b/api/v1_track_stream_test.go @@ -119,3 +119,62 @@ func TestGetTrackStream_DeletedTrack(t *testing.T) { assert.Equal(t, 404, res.StatusCode) assert.Empty(t, res.Header.Get("Location")) } + +// 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{ + "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, + }) +} + +// 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{ + "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 1fcdcf0f..4011169a 100644 --- a/indexer/indexer.go +++ b/indexer/indexer.go @@ -229,6 +229,12 @@ 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 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) } func (ci *CoreIndexer) Close() { diff --git a/jobs/repair_track_cids.go b/jobs/repair_track_cids.go new file mode 100644 index 00000000..e4d40f80 --- /dev/null +++ b/jobs/repair_track_cids.go @@ -0,0 +1,354 @@ +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 track_cid for tracks whose audio finished +// transcoding but were indexed without the cid. +// +// 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 +// 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 + + // 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 ( + // 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. 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 + // 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 { + 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, j.backedOffTrackIDs(time.Now())) + 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)) + } + if ok { + repaired++ + delete(j.retries, t.TrackID) + } else { + j.recordFailure(t.TrackID, time.Now()) + } + } + + 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 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, skip) + 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() +} + +// 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) { + 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 transcode not finished. + 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 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), + 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..4bf69ea8 --- /dev/null +++ b/jobs/repair_track_cids_test.go @@ -0,0 +1,272 @@ +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 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()) + + // 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 +} + +// A cidless track is repaired when nodes reach quorum. +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") +} + +// One node is not enough to set track_cid. +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)) +} + +// Disagreeing nodes leave the row unchanged. +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) +} + +// 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() + + 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"}, + {"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(), []int64{105}) + 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)) +} + +// 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") +} + +// applyTrackCid never overwrites an existing cid. +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) +}