mirror of
https://github.com/navidrome/navidrome.git
synced 2026-08-31 07:30:32 +00:00
feat(artwork): store the resolution trace so artwork explain works offline (#5980)
* feat(artwork): record the resolution trace so explain works without --live The worker never attached a ChainTrace, so `artwork explain` had to re-walk the priority chain at CLI time. That reconstruction could disagree with what actually happened, and without --live it could not report the external tier at all. The worker now traces every acquisition and stores it. `explain` reads the stored trace by default and reports when it was recorded; --live re-walks and calls the agents. Disc artwork keeps no row, so it always walks live. A chain trace alone would have explained almost nothing about failures: six of the seven ways an item can fail happen after the chain has already picked a winner. The trace now covers those stages too, and has somewhere to live when they fail: the retrying queue row carries the last failure, and the state row keeps it in last_failure once the retry budget is spent and the queue row is deleted. Measured on a copy of a 682MB / 43.6k-item library: +9.7MB (+1.4%). No row crosses the WITHOUT ROWID overflow threshold, so list hydration is unchanged; only full scans of item_artwork, which no request performs, read more pages. * test(artwork): pin the give-up ordering that keeps a failure for unresolved items recordGiveUp updates an existing row, and for a kind with a recheck path that row is only created moments earlier by the absent settle. Recording before the settle would lose the failure for every item that never resolved, with nothing to catch it. * refactor(artwork): tighten the trace code after review Four fixes worth taking: The doc comments on ChainTrace and chainState.trace still said the worker never attaches a trace and resolution stays allocation-free — the exact invariant this branch reverses. explain's report field meant both "the chain shown was walked just now" and "go out for real", and was being passed to loadPluginAgents, which --live documents as the only thing that may open external connections. Renamed to `walked` and restored explainLive as the sole input to that decision. A stored Detail is an error string on the failure paths, with no bound. The measured "no row reaches the WITHOUT ROWID overflow limit" only holds while it is bounded, so cap it at 200 runes. offlineGate was a factory returning a constant closure; make it a plain gateFunc like its sibling passthroughGate. Collapse five copies of the age-a-queue-row loop in the worker tests into one helper. * refactor(artwork): drop the offline explain walk, now that traces are stored `artwork explain` reported the external tier without calling it, so a diagnostic could not add load to a provider already rate-limiting us. Reading the stored trace answers that better: it reports what the agents actually returned, not what would be tried. Nothing could reach the offline gate any more. It was installed only for a walk with --live unset, which now happens for disc artwork alone, and disc rejects the external candidate before any gate call. That made the gate, its sentinel error, the would-try outcome and two of explain's verdicts unreachable. Removes offlineGate, errOfflineSkipped, OutcomeWouldTry, the NewTracingResolver live parameter and the CreateArtworkResolver argument threaded through wire. Verified against a copy of a real library: disc artwork with "external" first in DiscArtPriority and external services enabled still records the skip and issues no agent call. * fix(artwork): make explain's no-network guarantee structural, not incidental Serving falls back disc -> album and track -> disc -> album. The resolver layer explain uses has no such fallback today, so dropping the offline gate did not leak. But the guarantee rested on which chains happen to lack an external tier, and the serving layer already shows the fallback shape someone could mirror. Without --live the tracing resolver is now built with no agents at all, so no chain and no fallback added later can reach a provider. That is stronger than the gate it replaces, which only intercepted the call. The test pins it against exactly that regression: with the guard removed and the serving fallback mirrored into resolveDisc, it fails. * refactor(artwork): trim the trace plumbing EncodeTrace was exported for nobody: only this package writes traces, and cmd reads them. It becomes a ChainTrace method, which also drops the copy Steps made for a caller that only wanted to serialize. explain's report carried queuedSteps and failureSteps, both pure functions of the queue and state rows already in the struct, which let a test set the two out of step with each other. formatExplain derives them, as it already does for every other display value. The trace row format and its tabwriter empty-cell rule lived in two places, and the "nothing was ever recorded" predicate in three. * fix(artwork): clear the queue trace on a fresh re-enqueue Enqueue's conflict clause reset attempts to 0 but left the new trace column, so after a scan or refresh re-enqueued a previously-failed item artwork explain showed "Attempts: 0" next to the prior lifecycle's "Last attempt failed" trace. Clear trace in Enqueue (a fresh lifecycle has no last attempt); EnqueuePreservingBackoff still keeps it. * fix(artwork): treat a processing-stage error as indeterminate in explain A read/hash/decode/store failure records an OutcomeError step and writes an absent row, but explainResult only mapped external errors and unreadable candidates to indeterminate, so the default verdict read "not resolved" — presenting a processing failure as a definitive miss. The worker retries these exactly as it retries an unreadable candidate, so classify any OutcomeError as indeterminate too. * fix(artwork): record a trace step when a chainless resolver faults Playlist and radio resolvers walk no priority chain, so a fault (unreadable upload/sidecar, or an m3u fetch error with no grid) returned localError/extError without recording any trace step. The attempt then encoded [], leaving artwork explain with an empty "Last attempt failed" and "Gave up after". Record a fallback step in the faulted-no-image branch when nothing else did, and carry the source label through resolveLocalFile so the step can name it. * fix(artwork): trace the m3u failure at its source, not via the empty guard A playlist's grid sampling records album-chain steps into the shared trace, so the processor's empty-trace fallback no longer fires when the m3u remote image fetch failed — the error that forced the retry was omitted from explain. Record it where it happens, in resolvePlaylist's external step, as external:m3u. * test(artwork): skip the chainless-fault spec on Windows The spec provokes an open fault with a non-directory parent, but Windows maps that to a not-exist error, so localError is never set and the item resolves absent instead of failed. The sibling failed-on-unreadable-upload spec skips Windows for the same class of reason. * fix(artwork): don't label an absent empty-chain row as pre-tracing explain reported "resolved before traces were recorded" for any stored row with an empty chain, but an empty CoverArtPriority records a real, empty [] chain and resolves absent. A recorded resolution that finds an image always records its winning candidate, so only a row with a hash and no chain predates tracing; split on the hash and report an absent empty chain plainly instead. * fix(db): retimestamp the artwork trace migration after rebase master merged a 2026-08-18 migration, so the original 2026-08-16 timestamp is now older than the newest on the base branch and Goose would silently skip it on an already-upgraded database. Bumped past it; the SQL is unchanged. * fix(artwork): keep the m3u error detail in the trace The m3u trace step recorded OutcomeError with no detail because resolveExternalStep collapsed the gate's error to a bool, so explain showed only "external:m3u error -" and could not tell a timeout from an HTTP error or an open breaker. Return the error (normalizing not-found to nil so it stays a definitive miss, not a failure) and store its message as the step detail; encodeSteps already bounds it. * docs(artwork): note the give-up write relies on serial draining recordGiveUp writes last_failure unconditionally; that is only correct because the drain resolves each item serially, so no concurrent success can store artwork between the write and the queue delete. Record the invariant at the call site.
This commit is contained in:
parent
0a55cc8caf
commit
c26f6f9e98
108
cmd/artwork.go
108
cmd/artwork.go
@ -36,8 +36,9 @@ var (
|
||||
|
||||
func init() {
|
||||
artworkExplainCmd.Flags().BoolVar(&explainLive, "live", false,
|
||||
"perform real external lookups instead of reporting what would be tried; "+
|
||||
"also initializes plugin agents, which may open external connections")
|
||||
"walk the chain again now, performing real external lookups, instead of reporting the "+
|
||||
"stored trace of the last resolution; also initializes plugin agents, which may open "+
|
||||
"external connections")
|
||||
artworkReprocessCmd.Flags().StringSliceVar(&reprocessKinds, "kind", nil,
|
||||
"kinds to reprocess ("+kindPrefixes(artwork.RecheckKinds)+"); repeatable")
|
||||
artworkReprocessCmd.Flags().StringSliceVar(&reprocessSources, "source", nil,
|
||||
@ -640,10 +641,6 @@ func explainResult(source string, steps []artwork.TraceStep) string {
|
||||
if s.Outcome == artwork.OutcomeHit {
|
||||
break
|
||||
}
|
||||
if s.Outcome == artwork.OutcomeWouldTry {
|
||||
return "resolved from " + source +
|
||||
" (offline: a higher-priority external candidate was not tried; re-run with --live)"
|
||||
}
|
||||
// An external winner discards the earlier error, so the resolver settles it with no retry.
|
||||
if s.Outcome == artwork.OutcomeError && strings.HasPrefix(s.Candidate, artwork.ExternalPrefix) &&
|
||||
!strings.HasPrefix(source, artwork.ExternalPrefix) {
|
||||
@ -655,14 +652,12 @@ func explainResult(source string, steps []artwork.TraceStep) string {
|
||||
}
|
||||
for _, s := range steps {
|
||||
switch {
|
||||
case s.Outcome == artwork.OutcomeWouldTry:
|
||||
return "indeterminate (external agents not called; re-run with --live)"
|
||||
case s.Outcome == artwork.OutcomeError && strings.HasPrefix(s.Candidate, artwork.ExternalPrefix):
|
||||
return "indeterminate (an external lookup failed; the item may resolve on a later attempt)"
|
||||
// The worker treats an unreadable local candidate exactly as it treats a failed external one:
|
||||
// it retries instead of settling absent, so the verdict must not read as a clean miss.
|
||||
case s.Outcome == artwork.OutcomeUnreadable:
|
||||
return "indeterminate (a candidate exists but could not be read; the worker retries rather than settling absent)"
|
||||
// A stage error or an unreadable candidate means a source was found but not processed; the
|
||||
// worker retries rather than settling absent, so neither reads as a clean miss.
|
||||
case s.Outcome == artwork.OutcomeError, s.Outcome == artwork.OutcomeUnreadable:
|
||||
return "indeterminate (a candidate was found but could not be processed; the worker retries rather than settling absent)"
|
||||
}
|
||||
}
|
||||
return "not resolved"
|
||||
@ -684,22 +679,55 @@ func explainConfig(kind model.Kind) (name, value string) {
|
||||
}
|
||||
|
||||
type explainReport struct {
|
||||
kind model.Kind
|
||||
id string
|
||||
name string
|
||||
stored *model.ItemArtwork
|
||||
queued *model.ArtworkQueueItem
|
||||
agents string
|
||||
kind model.Kind
|
||||
id string
|
||||
name string
|
||||
stored *model.ItemArtwork
|
||||
queued *model.ArtworkQueueItem
|
||||
agents string
|
||||
// steps is the chain walk: recorded when the item was resolved, or performed just now when walked.
|
||||
steps []artwork.TraceStep
|
||||
source string
|
||||
walked bool
|
||||
resolveErr error
|
||||
}
|
||||
|
||||
// explainChainOrigin says whether the operator is reading history or a walk performed just now,
|
||||
// since the two can disagree after a config change.
|
||||
func explainChainOrigin(rep explainReport) string {
|
||||
if rep.walked {
|
||||
return "walked now"
|
||||
}
|
||||
if rep.stored != nil {
|
||||
return "recorded " + formatTime(rep.stored.AttemptedAt)
|
||||
}
|
||||
return "not recorded"
|
||||
}
|
||||
|
||||
// writeSteps prints the trace rows. An empty last cell would end tabwriter's column block and
|
||||
// break the alignment, so a missing detail is rendered as a dash.
|
||||
func writeSteps(w io.Writer, indent string, steps []artwork.TraceStep) {
|
||||
for _, s := range steps {
|
||||
fmt.Fprintf(w, "%s%s\t%s\t%s\n", indent, s.Candidate, s.Outcome, cmp.Or(s.Detail, "-"))
|
||||
}
|
||||
}
|
||||
|
||||
// writeStepTable prints a secondary trace, and nothing at all when there is none to show.
|
||||
func writeStepTable(w io.Writer, title string, steps []artwork.TraceStep) {
|
||||
if len(steps) == 0 {
|
||||
return
|
||||
}
|
||||
// No tab on the title: it closes the preceding column block, so these rows align among themselves.
|
||||
fmt.Fprintf(w, " %s:\n", title)
|
||||
writeSteps(w, " ", steps)
|
||||
}
|
||||
|
||||
func formatExplain(rep explainReport) string {
|
||||
var sb strings.Builder
|
||||
w := newTabWriter(&sb)
|
||||
explainable := artwork.Explainable(rep.kind)
|
||||
stateful := artwork.KeepsState(rep.kind)
|
||||
unrecorded := !rep.walked && rep.stored == nil
|
||||
|
||||
fmt.Fprintln(w, "Item")
|
||||
fmt.Fprintf(w, " Kind:\t%s (%s)\n", rep.kind, rep.kind.Prefix())
|
||||
@ -732,6 +760,12 @@ func formatExplain(rep explainReport) string {
|
||||
fmt.Fprintf(w, " Attempts:\t%d\n", rep.queued.Attempts)
|
||||
fmt.Fprintf(w, " Retry at:\t%s\n", formatTime(rep.queued.RetryAt))
|
||||
}
|
||||
if rep.queued != nil {
|
||||
writeStepTable(w, "Last attempt failed", artwork.DecodeTrace(rep.queued.Trace, ""))
|
||||
}
|
||||
if rep.stored != nil {
|
||||
writeStepTable(w, "Gave up after", artwork.DecodeTrace(rep.stored.LastFailure, ""))
|
||||
}
|
||||
|
||||
fmt.Fprintln(w, "\nConfig")
|
||||
if setting, value := explainConfig(rep.kind); setting == "" {
|
||||
@ -743,15 +777,22 @@ func formatExplain(rep explainReport) string {
|
||||
}
|
||||
}
|
||||
|
||||
fmt.Fprintln(w, "\nChain")
|
||||
if !explainable {
|
||||
fmt.Fprintf(w, "\nChain (%s)\n", explainChainOrigin(rep))
|
||||
switch {
|
||||
case !explainable:
|
||||
fmt.Fprintf(w, " (%s artwork does not walk a priority chain)\n", rep.kind)
|
||||
} else {
|
||||
case unrecorded:
|
||||
fmt.Fprintln(w, " (no resolution recorded yet; re-run with --live to walk the chain now)")
|
||||
case !rep.walked && len(rep.steps) == 0 && rep.stored.Hash != "":
|
||||
// A stored image with no chain can only predate trace recording: a recorded resolution that
|
||||
// found an image always records its winning candidate.
|
||||
fmt.Fprintln(w, " (this item was resolved before traces were recorded; re-run with --live)")
|
||||
case !rep.walked && len(rep.steps) == 0:
|
||||
// Absent with no chain: an empty priority list walked nothing, or a pre-tracing absent row.
|
||||
fmt.Fprintln(w, " (no candidates were recorded; re-run with --live to walk the chain now)")
|
||||
default:
|
||||
fmt.Fprintln(w, " CANDIDATE\tOUTCOME\tDETAIL")
|
||||
for _, s := range rep.steps {
|
||||
// A row with an empty last cell would end tabwriter's column block, breaking alignment.
|
||||
fmt.Fprintf(w, " %s\t%s\t%s\n", s.Candidate, s.Outcome, cmp.Or(s.Detail, "-"))
|
||||
}
|
||||
writeSteps(w, " ", rep.steps)
|
||||
}
|
||||
|
||||
fmt.Fprintln(w, "\nResult")
|
||||
@ -760,6 +801,8 @@ func formatExplain(rep explainReport) string {
|
||||
fmt.Fprintf(w, " resolution failed: %s\n", rep.resolveErr)
|
||||
case !explainable:
|
||||
fmt.Fprintln(w, " not evaluated (no chain was walked; see Stored above)")
|
||||
case unrecorded:
|
||||
fmt.Fprintln(w, " not evaluated (nothing recorded; re-run with --live to walk the chain now)")
|
||||
default:
|
||||
fmt.Fprintf(w, " %s\n", explainResult(rep.source, rep.steps))
|
||||
}
|
||||
@ -807,6 +850,8 @@ func runExplain(ctx context.Context, args []string) {
|
||||
}
|
||||
}
|
||||
|
||||
// Disc artwork keeps no row, so it has no stored trace and can only be explained by walking now.
|
||||
rep.walked = explainLive || !artwork.KeepsState(kind)
|
||||
if artwork.Explainable(kind) {
|
||||
// Only artist and album reach an agent, and the load must precede the resolver, which reads
|
||||
// the same manager.
|
||||
@ -815,11 +860,16 @@ func runExplain(ctx context.Context, args []string) {
|
||||
defer func() { _ = mgr.Stop() }()
|
||||
rep.agents = explainAgents(conf.Server.Agents, availableImageAgents(ds, mgr, kind))
|
||||
}
|
||||
trace := &artwork.ChainTrace{}
|
||||
rep.source, rep.resolveErr = CreateArtworkResolver(trace, explainLive).Resolve(ctx, kind, id)
|
||||
rep.steps = trace.Steps()
|
||||
switch {
|
||||
case rep.walked:
|
||||
trace := &artwork.ChainTrace{}
|
||||
rep.source, rep.resolveErr = CreateArtworkResolver(trace, explainLive).Resolve(ctx, kind, id)
|
||||
rep.steps = trace.Steps()
|
||||
case rep.stored != nil:
|
||||
rep.steps = artwork.DecodeTrace(rep.stored.Trace, rep.stored.SourcePath)
|
||||
rep.source = rep.stored.Source
|
||||
}
|
||||
}
|
||||
|
||||
fmt.Print(formatExplain(rep))
|
||||
// The steps taken before a failed walk are the diagnosis, so report them before exiting.
|
||||
if rep.resolveErr != nil {
|
||||
|
||||
@ -145,6 +145,15 @@ var _ = Describe("explainResult", func() {
|
||||
"the worker retries an unreadable candidate instead of settling absent, so this is not a clean miss")
|
||||
})
|
||||
|
||||
It("reports indeterminate when a processing stage errored after a candidate was found", func() {
|
||||
steps := []artwork.TraceStep{
|
||||
{Candidate: "cover.*", Outcome: "hit", Detail: "/music/cover.jpg"},
|
||||
{Candidate: "store", Outcome: "error", Detail: "disk full"},
|
||||
}
|
||||
Expect(explainResult("", steps)).To(ContainSubstring("indeterminate"),
|
||||
"a stage error is a processing failure the worker retries, not a definitive miss")
|
||||
})
|
||||
|
||||
It("does not qualify a hit that an earlier unreadable candidate preceded", func() {
|
||||
// chainState.try stamps only the external error onto a hit and drops the local one, so the
|
||||
// worker settles this as found; warning about it would be a false alarm.
|
||||
@ -164,34 +173,6 @@ var _ = Describe("explainResult", func() {
|
||||
"a failed network call is not evidence that the item has no artwork")
|
||||
})
|
||||
|
||||
It("qualifies a win a skipped higher-priority external candidate could have taken", func() {
|
||||
steps := []artwork.TraceStep{
|
||||
{Candidate: "external:deezer", Outcome: "would-try"},
|
||||
{Candidate: "artist.*", Outcome: "hit", Detail: "/music/artist.jpg"},
|
||||
}
|
||||
res := explainResult("artist.*", steps)
|
||||
Expect(res).To(ContainSubstring("resolved from artist.*"))
|
||||
Expect(res).To(ContainSubstring("--live"),
|
||||
"offline, the winner is only the winner because the external tier was skipped")
|
||||
})
|
||||
|
||||
It("does not qualify a win that no skipped candidate outranked", func() {
|
||||
steps := []artwork.TraceStep{
|
||||
{Candidate: "artist.*", Outcome: "hit"},
|
||||
{Candidate: "external:deezer", Outcome: "would-try"},
|
||||
}
|
||||
Expect(explainResult("artist.*", steps)).To(Equal("resolved from artist.*"))
|
||||
})
|
||||
|
||||
It("reports indeterminate when external agents were never called", func() {
|
||||
steps := []artwork.TraceStep{
|
||||
{Candidate: "artist.*", Outcome: "miss"},
|
||||
{Candidate: "external:deezer", Outcome: "would-try"},
|
||||
}
|
||||
Expect(explainResult("", steps)).To(ContainSubstring("indeterminate"),
|
||||
"an offline run must not claim an item is unresolvable when external agents were skipped")
|
||||
})
|
||||
|
||||
It("qualifies a win a failed higher-priority external lookup could have taken", func() {
|
||||
steps := []artwork.TraceStep{
|
||||
{Candidate: "external:deezer", Outcome: "error", Detail: "context deadline exceeded"},
|
||||
@ -252,9 +233,10 @@ var _ = Describe("formatExplain", func() {
|
||||
id: "ar-1",
|
||||
name: "Radiohead",
|
||||
agents: "lastfm,spotify",
|
||||
walked: true,
|
||||
steps: []artwork.TraceStep{
|
||||
{Candidate: "upload", Outcome: "skipped", Detail: "no uploaded image"},
|
||||
{Candidate: "external:deezer", Outcome: "would-try"},
|
||||
{Candidate: "external:deezer", Outcome: "error", Detail: "context deadline exceeded"},
|
||||
},
|
||||
source: "",
|
||||
}
|
||||
@ -267,7 +249,6 @@ var _ = Describe("formatExplain", func() {
|
||||
Expect(out).To(ContainSubstring("ArtistArtPriority"))
|
||||
Expect(out).To(ContainSubstring("lastfm,spotify"))
|
||||
Expect(out).To(ContainSubstring("external:deezer"))
|
||||
Expect(out).To(ContainSubstring("would-try"))
|
||||
Expect(out).To(ContainSubstring("indeterminate"))
|
||||
})
|
||||
|
||||
@ -304,7 +285,7 @@ var _ = Describe("formatExplain", func() {
|
||||
out := formatExplain(rep)
|
||||
Expect(out).To(ContainSubstring("resolution failed: no such directory"))
|
||||
Expect(out).ToNot(ContainSubstring("indeterminate"))
|
||||
Expect(out).To(ContainSubstring("would-try"), "the steps taken before the failure still print")
|
||||
Expect(out).To(ContainSubstring("external:deezer"), "the steps taken before the failure still print")
|
||||
})
|
||||
|
||||
It("says a kind that does not walk a chain has no chain, without an empty table", func() {
|
||||
@ -329,6 +310,7 @@ var _ = Describe("formatExplain", func() {
|
||||
kind: model.KindDiscArtwork, id: "al-1:2", name: "OK Computer (disc 2)",
|
||||
steps: []artwork.TraceStep{{Candidate: "cover.jpg", Outcome: "hit", Detail: "/music/cover.jpg"}},
|
||||
source: "folder",
|
||||
walked: true,
|
||||
}
|
||||
|
||||
out := formatExplain(rep)
|
||||
@ -341,10 +323,74 @@ var _ = Describe("formatExplain", func() {
|
||||
Expect(out).To(ContainSubstring("resolved from folder"))
|
||||
})
|
||||
|
||||
Context("stored traces", func() {
|
||||
BeforeEach(func() {
|
||||
rep.walked = false
|
||||
rep.steps = nil
|
||||
})
|
||||
|
||||
It("labels a recorded chain with when it was recorded, not as a walk done now", func() {
|
||||
attempted := time.Date(2026, 8, 13, 10, 0, 0, 0, time.UTC)
|
||||
rep.stored = &model.ItemArtwork{Source: "folder", Hash: "abc", AttemptedAt: attempted}
|
||||
rep.steps = []artwork.TraceStep{{Candidate: "artist.*", Outcome: "hit", Detail: "/music/artist.jpg"}}
|
||||
rep.source = "folder"
|
||||
|
||||
out := formatExplain(rep)
|
||||
Expect(out).To(ContainSubstring("Chain (recorded 2026-08-13T10:00:00Z)"))
|
||||
Expect(out).To(ContainSubstring("/music/artist.jpg"))
|
||||
Expect(out).To(ContainSubstring("resolved from folder"))
|
||||
})
|
||||
|
||||
It("says so when the item has never been resolved", func() {
|
||||
out := formatExplain(rep)
|
||||
Expect(out).To(ContainSubstring("no resolution recorded yet"))
|
||||
Expect(out).To(ContainSubstring("--live"))
|
||||
Expect(out).ToNot(ContainSubstring("not resolved"),
|
||||
"nothing was recorded, which is not the same as resolving to nothing")
|
||||
})
|
||||
|
||||
It("distinguishes a row written before traces existed from one with an empty chain", func() {
|
||||
rep.stored = &model.ItemArtwork{Source: "folder", Hash: "abc", AttemptedAt: time.Now()}
|
||||
|
||||
Expect(formatExplain(rep)).To(ContainSubstring("resolved before traces were recorded"))
|
||||
})
|
||||
|
||||
It("does not call an absent row with an empty recorded chain a pre-tracing row", func() {
|
||||
// An empty priority list records a real but empty chain and resolves absent; that is not a
|
||||
// legacy row, so it must not be reported as resolved before tracing existed.
|
||||
rep.stored = &model.ItemArtwork{Source: "", Hash: "", AttemptedAt: time.Now()}
|
||||
|
||||
out := formatExplain(rep)
|
||||
Expect(out).ToNot(ContainSubstring("resolved before traces were recorded"))
|
||||
Expect(out).To(ContainSubstring("no candidates were recorded"))
|
||||
Expect(out).To(ContainSubstring("not resolved"), "the Result still reports the absence plainly")
|
||||
})
|
||||
|
||||
It("prints why the last attempt failed and why it gave up", func() {
|
||||
rep.queued = &model.ArtworkQueueItem{Priority: model.ArtworkPriorityScan, Attempts: 3,
|
||||
Trace: `[{"c":"decode","o":"error","d":"bad header"}]`}
|
||||
rep.stored = &model.ItemArtwork{Source: "folder", Hash: "abc", AttemptedAt: time.Now(),
|
||||
LastFailure: `[{"c":"read","o":"error","d":"i/o timeout"}]`}
|
||||
|
||||
out := formatExplain(rep)
|
||||
Expect(out).To(ContainSubstring("Last attempt failed"))
|
||||
Expect(out).To(ContainSubstring("bad header"))
|
||||
Expect(out).To(ContainSubstring("Gave up after"))
|
||||
Expect(out).To(ContainSubstring("i/o timeout"))
|
||||
})
|
||||
|
||||
It("omits the failure tables when there is no failure to report", func() {
|
||||
out := formatExplain(rep)
|
||||
Expect(out).ToNot(ContainSubstring("Last attempt failed"))
|
||||
Expect(out).ToNot(ContainSubstring("Gave up after"))
|
||||
})
|
||||
})
|
||||
|
||||
It("reports the setting that governs media file artwork", func() {
|
||||
conf.Server.EnableMediaFileCoverArt = false
|
||||
rep = explainReport{
|
||||
kind: model.KindMediaFileArtwork, id: "mf-1", name: "Airbag",
|
||||
walked: true,
|
||||
steps: []artwork.TraceStep{
|
||||
{Candidate: "embedded", Outcome: "skipped", Detail: "EnableMediaFileCoverArt is off"},
|
||||
},
|
||||
|
||||
@ -58,7 +58,7 @@ func fetchArtistImage(ctx context.Context, ag *agents.Agents, gate gateFunc, ar
|
||||
return nil, "", false
|
||||
}
|
||||
for _, a := range imageAgents {
|
||||
reader, _, err := gate(a.Name, func() (io.ReadCloser, string, error) {
|
||||
reader, path, err := gate(a.Name, func() (io.ReadCloser, string, error) {
|
||||
imgs, err := a.Retriever.GetArtistImages(ctx, ar.ID, name, ar.MbzArtistID)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
@ -69,6 +69,7 @@ func fetchArtistImage(ctx context.Context, ag *agents.Agents, gate gateFunc, ar
|
||||
}
|
||||
return fromURL(ctx, u)
|
||||
})
|
||||
recordAgent(ctx, a.Name, reader, path, err)
|
||||
if reader != nil {
|
||||
return reader, a.Name, false
|
||||
}
|
||||
@ -90,7 +91,7 @@ func fetchAlbumImage(ctx context.Context, ag *agents.Agents, gate gateFunc, al m
|
||||
return nil, "", false
|
||||
}
|
||||
for _, a := range imageAgents {
|
||||
reader, _, err := gate(a.Name, func() (io.ReadCloser, string, error) {
|
||||
reader, path, err := gate(a.Name, func() (io.ReadCloser, string, error) {
|
||||
imgs, err := a.Retriever.GetAlbumImages(ctx, name, artist, al.MbzAlbumID)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
@ -101,6 +102,7 @@ func fetchAlbumImage(ctx context.Context, ag *agents.Agents, gate gateFunc, al m
|
||||
}
|
||||
return fromURL(ctx, u)
|
||||
})
|
||||
recordAgent(ctx, a.Name, reader, path, err)
|
||||
if reader != nil {
|
||||
return reader, a.Name, false
|
||||
}
|
||||
|
||||
@ -393,16 +393,15 @@ type TracingResolver struct {
|
||||
trace *ChainTrace
|
||||
}
|
||||
|
||||
// NewTracingResolver builds a TracingResolver that records its priority-chain walk. With live
|
||||
// false the external tier is reported but never called.
|
||||
// NewTracingResolver builds a TracingResolver that records its priority-chain walk. Without live
|
||||
// it gets no agents at all, so neither a chain nor any fallback added later can reach a provider;
|
||||
// with it, one item is at most one call per agent, so the rate limiter and breaker are bypassed.
|
||||
func NewTracingResolver(ds model.DataStore, ag *agents.Agents, ffm ffmpeg.FFmpeg, t *ChainTrace, live bool) *TracingResolver {
|
||||
gate := offlineGate(t)
|
||||
inner := newLocalResolver(ds, ffm)
|
||||
if live {
|
||||
// A diagnostic must show the provider's real answer, and one item is at most one call
|
||||
// per agent, so --live deliberately bypasses the rate limiter and circuit breaker.
|
||||
gate = tracingGate(t, passthroughGate)
|
||||
inner = newResolver(ds, ag, ffm, passthroughGate)
|
||||
}
|
||||
return &TracingResolver{inner: newResolver(ds, ag, ffm, gate), trace: t}
|
||||
return &TracingResolver{inner: inner, trace: t}
|
||||
}
|
||||
|
||||
// Resolve walks kind's sources for id, recording the walk, and reports the winning source
|
||||
|
||||
@ -2,6 +2,7 @@ package artwork
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"cmp"
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"errors"
|
||||
@ -89,12 +90,21 @@ func (p *processor) acquire(ctx context.Context, item model.ArtworkQueueItem) (o
|
||||
|
||||
res, err := p.resolver.resolve(ctx, item)
|
||||
if err != nil {
|
||||
traceStage(ctx, "resolve", err)
|
||||
log.Warn(ctx, "Artwork: Could not resolve item", "kind", item.ItemKind, "id", item.ItemID, err)
|
||||
return outcomeFailed, nil
|
||||
}
|
||||
if res.reader == nil {
|
||||
if res.extError || res.localError {
|
||||
// A fault is not a definitive "no image": never settle absent, keep serving old state.
|
||||
// A chainless resolver (playlist/radio) records no step, so leave a fallback or explain is blank.
|
||||
if t := traceFrom(ctx); len(t.Steps()) == 0 {
|
||||
outcome := OutcomeError
|
||||
if res.localError {
|
||||
outcome = OutcomeUnreadable
|
||||
}
|
||||
t.add(TraceStep{Candidate: cmp.Or(res.source, "source"), Outcome: outcome})
|
||||
}
|
||||
log.Debug(ctx, "Artwork: No image, but a source faulted; keeping previous state",
|
||||
"kind", item.ItemKind, "id", item.ItemID, "extError", res.extError, "localError", res.localError)
|
||||
return outcomeFailed, nil
|
||||
@ -106,6 +116,7 @@ func (p *processor) acquire(ctx context.Context, item model.ArtworkQueueItem) (o
|
||||
readStart := time.Now()
|
||||
data, err := readCapped(res.reader)
|
||||
if err != nil {
|
||||
traceStage(ctx, "read", err)
|
||||
log.Warn(ctx, "Artwork: Failed to read resolved image", "kind", item.ItemKind, "id", item.ItemID, "source", res.source, err)
|
||||
return outcomeFailed, nil
|
||||
}
|
||||
@ -115,6 +126,7 @@ func (p *processor) acquire(ctx context.Context, item model.ArtworkQueueItem) (o
|
||||
hashStart := time.Now()
|
||||
hash, err := hashImage(bytes.NewReader(data))
|
||||
if err != nil {
|
||||
traceStage(ctx, "hash", err)
|
||||
log.Warn(ctx, "Artwork: Failed to hash image", "kind", item.ItemKind, "id", item.ItemID, err)
|
||||
return outcomeFailed, nil
|
||||
}
|
||||
@ -138,19 +150,22 @@ func (p *processor) acquire(ctx context.Context, item model.ArtworkQueueItem) (o
|
||||
art, err = undecodedArtwork(hash), nil
|
||||
}
|
||||
if err != nil {
|
||||
traceStage(ctx, "decode", err)
|
||||
log.Warn(ctx, "Artwork: Failed to decode resolved image", "kind", item.ItemKind, "id", item.ItemID, err)
|
||||
return outcomeFailed, nil
|
||||
}
|
||||
log.Debug(ctx, "Artwork: Decoded new image", "kind", item.ItemKind, "id", item.ItemID, "hash", hash,
|
||||
"width", art.Width, "height", art.Height, "mime", art.Mime, "elapsed", time.Since(decodeStart))
|
||||
default:
|
||||
traceStage(ctx, "lookup", err)
|
||||
log.Warn(ctx, "Artwork: Failed to look up image hash", "kind", item.ItemKind, "id", item.ItemID, err)
|
||||
return outcomeFailed, nil
|
||||
}
|
||||
art.SizeBytes = int64(len(data))
|
||||
|
||||
ia, err := p.persist(repo, item, art, res, data)
|
||||
ia, err := p.persist(ctx, repo, item, art, res, data)
|
||||
if err != nil {
|
||||
traceStage(ctx, "store", err)
|
||||
log.Warn(ctx, "Artwork: Failed to persist resolved image", "kind", item.ItemKind, "id", item.ItemID, err)
|
||||
return outcomeFailed, nil
|
||||
}
|
||||
@ -165,7 +180,7 @@ func (p *processor) acquire(ctx context.Context, item model.ArtworkQueueItem) (o
|
||||
|
||||
// persist places the bytes and commits the rows referencing them, excluding Prune for that
|
||||
// window only so a slow resolution can never hold it off.
|
||||
func (p *processor) persist(repo model.ArtworkRepository, item model.ArtworkQueueItem,
|
||||
func (p *processor) persist(ctx context.Context, repo model.ArtworkRepository, item model.ArtworkQueueItem,
|
||||
art *model.Artwork, res resolution, data []byte,
|
||||
) (*model.ItemArtwork, error) {
|
||||
if p.pruneLock != nil {
|
||||
@ -188,6 +203,7 @@ func (p *processor) persist(repo model.ArtworkRepository, item model.ArtworkQueu
|
||||
SourcePath: sourcePath,
|
||||
RefMtime: refMtime,
|
||||
AttemptedAt: time.Now(),
|
||||
Trace: traceFrom(ctx).encode(sourcePath),
|
||||
}
|
||||
// PutItemArtwork stamps UpdatedAt on ia, so the returned struct matches the persisted row.
|
||||
if err := repo.PutItemArtwork(ia); err != nil {
|
||||
@ -203,6 +219,7 @@ func writeAbsent(ctx context.Context, repo model.ArtworkRepository, item model.A
|
||||
ItemID: item.ItemID,
|
||||
ImageType: item.ImageType,
|
||||
AttemptedAt: time.Now(),
|
||||
Trace: traceFrom(ctx).encode(""),
|
||||
})
|
||||
if err != nil {
|
||||
log.Warn(ctx, "Artwork: Failed to persist absent state", "kind", item.ItemKind, "id", item.ItemID, err)
|
||||
|
||||
@ -229,6 +229,35 @@ var _ = Describe("processor.acquire", func() {
|
||||
Expect(err).To(MatchError(model.ErrNotFound), "an unreadable upload must not be recorded as absent")
|
||||
})
|
||||
|
||||
// Playlist/radio resolvers walk no chain, so a fault records no step; without a fallback,
|
||||
// explain would show a give-up with an empty "Gave up after" table.
|
||||
It("chainless fault: records a fallback trace step naming the faulted source", func() {
|
||||
if runtime.GOOS == "windows" {
|
||||
// os.Open under a non-directory maps to a not-exist error on Windows, so no localError.
|
||||
Skip("cannot provoke an open fault via a non-directory parent on Windows")
|
||||
}
|
||||
radioRepo := tests.CreateMockedRadioRepo()
|
||||
radioRepo.Data = map[string]*model.Radio{}
|
||||
ds.MockedRadio = radioRepo
|
||||
dir := GinkgoT().TempDir()
|
||||
conf.Server.DataFolder = conf.NewDir(dir)
|
||||
upload := model.UploadedImagePath(consts.EntityRadio, "ra-tr.jpg")
|
||||
// A plain file where the upload's parent should be makes os.Open fault with ENOTDIR,
|
||||
// deterministically and regardless of the test user's privileges.
|
||||
Expect(os.MkdirAll(filepath.Dir(filepath.Dir(upload)), 0o755)).To(Succeed())
|
||||
Expect(os.WriteFile(filepath.Dir(upload), []byte("x"), 0o600)).To(Succeed())
|
||||
radioRepo.Data["ra-tr"] = &model.Radio{ID: "ra-tr", Name: "Station", UploadedImage: "ra-tr.jpg"}
|
||||
|
||||
trace := &ChainTrace{}
|
||||
out, _ := proc.acquire(withTrace(ctx, trace), model.ArtworkQueueItem{ItemKind: "ra", ItemID: "ra-tr"})
|
||||
Expect(out).To(Equal(outcomeFailed))
|
||||
|
||||
steps := trace.Steps()
|
||||
Expect(steps).To(HaveLen(1), "a radio fault must leave one step so explain is not blank")
|
||||
Expect(steps[0].Candidate).To(Equal("upload"))
|
||||
Expect(steps[0].Outcome).To(Equal(OutcomeUnreadable))
|
||||
})
|
||||
|
||||
It("failed-on-extError: leaves the item's state untouched", func() {
|
||||
conf.Server.CoverArtPriority = "external"
|
||||
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{
|
||||
|
||||
@ -37,7 +37,7 @@ type resolution struct {
|
||||
// transient external failure still retries; localErr is dropped, as the scanner re-lists changes.
|
||||
type chainState struct {
|
||||
extErr, localErr bool
|
||||
trace *ChainTrace // nil unless the CLI asked for a trace
|
||||
trace *ChainTrace // nil only where no caller attached one
|
||||
}
|
||||
|
||||
// try stamps the accumulated external failure onto a hit, and records the miss otherwise.
|
||||
@ -354,10 +354,13 @@ func (r *resolver) resolvePlaylist(ctx context.Context, playlistID string) (reso
|
||||
}
|
||||
if remoteImg != nil && conf.Server.EnableM3UExternalAlbumArt {
|
||||
sf := func() (io.ReadCloser, string, error) { return fromURL(ctx, remoteImg) }
|
||||
if res, ok, isErr := resolveExternalStep(r.ext.gate, "m3u", sf); ok {
|
||||
if res, ok, err := resolveExternalStep(r.ext.gate, "m3u", sf); ok {
|
||||
return res, nil
|
||||
} else if isErr {
|
||||
} else if err != nil {
|
||||
extErr = true
|
||||
// Record it here with its detail: once album sampling adds its own steps, the processor's
|
||||
// empty-trace fallback no longer fires, and the error that forced the retry would be lost.
|
||||
traceFrom(ctx).add(TraceStep{Candidate: ExternalPrefix + "m3u", Outcome: OutcomeError, Detail: err.Error()})
|
||||
}
|
||||
}
|
||||
|
||||
@ -461,14 +464,17 @@ func (r *resolver) resolveDisc(ctx context.Context, id string) (resolution, erro
|
||||
return dr.selectImage(ctx, r.ffmpeg, conf.Server.DiscArtPriority, &chain)
|
||||
}
|
||||
|
||||
// resolveExternalStep runs a single external sourceFunc through the named gate. extErr excludes
|
||||
// a not-found, which is a definitive "no" rather than a failure.
|
||||
func resolveExternalStep(gate gateFunc, name string, sf sourceFunc) (res resolution, ok bool, extErr bool) {
|
||||
// resolveExternalStep runs a single external sourceFunc through the named gate. A not-found is a
|
||||
// definitive "no", returned as (_, false, nil); any other error is a failure the caller records.
|
||||
func resolveExternalStep(gate gateFunc, name string, sf sourceFunc) (resolution, bool, error) {
|
||||
r, path, err := gate(name, sf)
|
||||
if r != nil {
|
||||
return resolution{reader: r, source: externalCandidate, sourcePath: path}, true, false
|
||||
return resolution{reader: r, source: externalCandidate, sourcePath: path}, true, nil
|
||||
}
|
||||
return resolution{}, false, err != nil && !errors.Is(err, model.ErrNotFound)
|
||||
if errors.Is(err, model.ErrNotFound) {
|
||||
return resolution{}, false, nil
|
||||
}
|
||||
return resolution{}, false, err
|
||||
}
|
||||
|
||||
// classifyPlaylistImage splits a playlist ExternalImageURL into a local filesystem path or a
|
||||
@ -561,7 +567,9 @@ func resolveLocalFile(path, source string) (resolution, bool) {
|
||||
}
|
||||
f, err := os.Open(path)
|
||||
if err != nil {
|
||||
return resolution{localError: !errors.Is(err, fs.ErrNotExist)}, false
|
||||
// Carry the source label even on a fault, so a resolver with no chain (playlist/radio) can
|
||||
// still name what faulted in the trace.
|
||||
return resolution{source: source, localError: !errors.Is(err, fs.ErrNotExist)}, false
|
||||
}
|
||||
return resolution{reader: f, source: source, sourcePath: path, refMtime: mtimeOf(path)}, true
|
||||
}
|
||||
|
||||
@ -520,6 +520,37 @@ var _ = Describe("resolveItem", func() {
|
||||
Expect(gatedNames).To(Equal([]string{"m3u"}), "the playlist URL fetch is gated under \"m3u\"")
|
||||
})
|
||||
|
||||
It("records the m3u failure in the trace even when album sampling adds its own steps", func() {
|
||||
conf.Server.EnableM3UExternalAlbumArt = true
|
||||
folderRepo.result = nil // the sampled album yields no tile, so the m3u failure is what forced the retry
|
||||
|
||||
plRepo := tests.CreateMockPlaylistRepo()
|
||||
plRepo.SetData(model.Playlists{{ID: "plm3u", Name: "Playlist", ExternalImageURL: "http://example.com/cover.jpg"}})
|
||||
plRepo.TracksRepo = &tests.MockPlaylistTrackRepo{AlbumIDs: []string{"t1"}}
|
||||
ds.MockedPlaylist = plRepo
|
||||
|
||||
gate := func(string, func() (io.ReadCloser, string, error)) (io.ReadCloser, string, error) {
|
||||
return nil, "", errors.New("network down")
|
||||
}
|
||||
|
||||
trace := &ChainTrace{}
|
||||
res, err := newResolver(ds, ag, ffm, gate).resolve(withTrace(ctx, trace),
|
||||
model.ArtworkQueueItem{ItemKind: "pl", ItemID: "plm3u"})
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.extError).To(BeTrue())
|
||||
|
||||
steps := trace.Steps()
|
||||
var m3u *TraceStep
|
||||
for i := range steps {
|
||||
if steps[i].Candidate == ExternalPrefix+"m3u" && steps[i].Outcome == OutcomeError {
|
||||
m3u = &steps[i]
|
||||
}
|
||||
}
|
||||
Expect(m3u).ToNot(BeNil(), "the m3u fetch error must be traced at its source, not left to the empty-trace fallback")
|
||||
Expect(m3u.Detail).To(Equal("network down"),
|
||||
"the trace must carry the underlying error so explain can tell a timeout from an HTTP error")
|
||||
})
|
||||
|
||||
It("treats a missing local ExternalImageURL as a definitive miss, not extError", func() {
|
||||
folderRepo.result = nil // no grid tiles, so the local-file miss is what surfaces
|
||||
|
||||
|
||||
@ -2,10 +2,12 @@ package artwork
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"slices"
|
||||
"sync"
|
||||
|
||||
"github.com/navidrome/navidrome/utils/str"
|
||||
)
|
||||
|
||||
// Outcome is what the priority chain observed for one candidate; the CLI renders and branches on these.
|
||||
@ -16,7 +18,6 @@ const (
|
||||
OutcomeMiss Outcome = "miss"
|
||||
OutcomeUnreadable Outcome = "unreadable"
|
||||
OutcomeSkipped Outcome = "skipped"
|
||||
OutcomeWouldTry Outcome = "would-try"
|
||||
OutcomeError Outcome = "error"
|
||||
)
|
||||
|
||||
@ -34,8 +35,8 @@ type TraceStep struct {
|
||||
Detail string
|
||||
}
|
||||
|
||||
// ChainTrace collects the walk of a single resolution. The artwork worker never attaches
|
||||
// one; only the CLI does, so resolution stays allocation-free in the hot path.
|
||||
// ChainTrace collects the walk of a single resolution: the worker attaches one per queue
|
||||
// item so it can be stored, and the CLI attaches one per explain.
|
||||
type ChainTrace struct {
|
||||
mu sync.Mutex
|
||||
steps []TraceStep
|
||||
@ -59,6 +60,65 @@ func (t *ChainTrace) Steps() []TraceStep {
|
||||
return slices.Clone(t.steps)
|
||||
}
|
||||
|
||||
// maxTraceDetail bounds a stored Detail, which on the failure paths is an error string of
|
||||
// unknown length. Past ~1kB a row spills to an overflow page, slowing every scan of the table.
|
||||
const maxTraceDetail = 200
|
||||
|
||||
// storedStep is the persisted shape of a TraceStep. The keys are single letters because a trace
|
||||
// is written for every item, and the encoded length is repeated across the whole library.
|
||||
type storedStep struct {
|
||||
C string `json:"c"`
|
||||
O Outcome `json:"o"`
|
||||
D string `json:"d,omitempty"`
|
||||
}
|
||||
|
||||
// encode serializes the trace for storage, without the copy Steps would make for a caller
|
||||
// that only wants to write it.
|
||||
func (t *ChainTrace) encode(sourcePath string) string {
|
||||
if t == nil {
|
||||
return encodeSteps(nil, sourcePath)
|
||||
}
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
return encodeSteps(t.steps, sourcePath)
|
||||
}
|
||||
|
||||
// encodeSteps writes the stored form. A hit's Detail is the winning source's path, which the
|
||||
// same row already stores as source_path, so it is dropped and DecodeTrace puts it back.
|
||||
func encodeSteps(steps []TraceStep, sourcePath string) string {
|
||||
out := make([]storedStep, 0, len(steps))
|
||||
for _, s := range steps {
|
||||
d := s.Detail
|
||||
if s.Outcome == OutcomeHit && d == sourcePath {
|
||||
d = ""
|
||||
}
|
||||
out = append(out, storedStep{C: s.Candidate, O: s.Outcome, D: str.TruncateRunes(d, maxTraceDetail, "...")})
|
||||
}
|
||||
b, _ := json.Marshal(out) // []storedStep is all strings, so this cannot fail
|
||||
return string(b)
|
||||
}
|
||||
|
||||
// DecodeTrace reverses the stored form. A trace that will not parse is reported as no trace at all,
|
||||
// since a diagnostic command must not fail on a bad row.
|
||||
func DecodeTrace(encoded, sourcePath string) []TraceStep {
|
||||
if encoded == "" {
|
||||
return nil
|
||||
}
|
||||
var stored []storedStep
|
||||
if err := json.Unmarshal([]byte(encoded), &stored); err != nil {
|
||||
return nil
|
||||
}
|
||||
steps := make([]TraceStep, 0, len(stored))
|
||||
for _, s := range stored {
|
||||
d := s.D
|
||||
if d == "" && s.O == OutcomeHit {
|
||||
d = sourcePath
|
||||
}
|
||||
steps = append(steps, TraceStep{Candidate: s.C, Outcome: s.O, Detail: d})
|
||||
}
|
||||
return steps
|
||||
}
|
||||
|
||||
type traceCtxKey struct{}
|
||||
|
||||
func withTrace(ctx context.Context, t *ChainTrace) context.Context {
|
||||
@ -70,30 +130,23 @@ func traceFrom(ctx context.Context) *ChainTrace {
|
||||
return t
|
||||
}
|
||||
|
||||
var errOfflineSkipped = errors.New("artwork: external lookup skipped (offline)")
|
||||
|
||||
// tracingGate records each external agent's outcome without changing what the gate returns.
|
||||
func tracingGate(t *ChainTrace, inner gateFunc) gateFunc {
|
||||
return func(name string, f func() (io.ReadCloser, string, error)) (io.ReadCloser, string, error) {
|
||||
r, path, err := inner(name, f)
|
||||
candidate := ExternalPrefix + name
|
||||
switch {
|
||||
case r != nil:
|
||||
t.add(TraceStep{Candidate: candidate, Outcome: OutcomeHit, Detail: path})
|
||||
case isTransientExternal(err):
|
||||
t.add(TraceStep{Candidate: candidate, Outcome: OutcomeError, Detail: err.Error()})
|
||||
default:
|
||||
t.add(TraceStep{Candidate: candidate, Outcome: OutcomeMiss})
|
||||
}
|
||||
return r, path, err
|
||||
// recordAgent files what one external agent answered. The agent loops call this rather than a
|
||||
// gate wrapper, because only they hold the context that carries the trace.
|
||||
func recordAgent(ctx context.Context, name string, r io.ReadCloser, path string, err error) {
|
||||
t := traceFrom(ctx)
|
||||
candidate := ExternalPrefix + name
|
||||
switch {
|
||||
case r != nil:
|
||||
t.add(TraceStep{Candidate: candidate, Outcome: OutcomeHit, Detail: path})
|
||||
case isTransientExternal(err):
|
||||
t.add(TraceStep{Candidate: candidate, Outcome: OutcomeError, Detail: err.Error()})
|
||||
default:
|
||||
t.add(TraceStep{Candidate: candidate, Outcome: OutcomeMiss})
|
||||
}
|
||||
}
|
||||
|
||||
// offlineGate reports which agents would be asked without asking them, so a diagnostic
|
||||
// command cannot add load to a provider that is already rate-limiting us.
|
||||
func offlineGate(t *ChainTrace) gateFunc {
|
||||
return func(name string, _ func() (io.ReadCloser, string, error)) (io.ReadCloser, string, error) {
|
||||
t.add(TraceStep{Candidate: ExternalPrefix + name, Outcome: OutcomeWouldTry})
|
||||
return nil, "", errOfflineSkipped
|
||||
}
|
||||
// traceStage records a failure from the stages that run after the priority chain has already
|
||||
// picked a winner: most ways an item can fail are here, not in the chain walk.
|
||||
func traceStage(ctx context.Context, stage string, err error) {
|
||||
traceFrom(ctx).add(TraceStep{Candidate: stage, Outcome: OutcomeError, Detail: err.Error()})
|
||||
}
|
||||
|
||||
@ -24,13 +24,63 @@ var _ = Describe("trace vocabulary", func() {
|
||||
// what `artwork explain` tells an operator, so it must be made deliberately.
|
||||
It("pins the wire values the CLI reads", func() {
|
||||
Expect([]Outcome{
|
||||
OutcomeHit, OutcomeMiss, OutcomeUnreadable, OutcomeSkipped, OutcomeWouldTry, OutcomeError,
|
||||
}).To(Equal([]Outcome{"hit", "miss", "unreadable", "skipped", "would-try", "error"}))
|
||||
OutcomeHit, OutcomeMiss, OutcomeUnreadable, OutcomeSkipped, OutcomeError,
|
||||
}).To(Equal([]Outcome{"hit", "miss", "unreadable", "skipped", "error"}))
|
||||
Expect(externalCandidate).To(Equal("external"))
|
||||
Expect(ExternalPrefix).To(Equal("external:"))
|
||||
})
|
||||
})
|
||||
|
||||
var _ = Describe("encodeSteps/DecodeTrace", func() {
|
||||
It("round-trips a trace", func() {
|
||||
steps := []TraceStep{
|
||||
{Candidate: "cover.png", Outcome: OutcomeMiss},
|
||||
{Candidate: "cover.*", Outcome: OutcomeHit, Detail: "/music/a/cover.jpg"},
|
||||
}
|
||||
Expect(DecodeTrace(encodeSteps(steps, ""), "")).To(Equal(steps))
|
||||
})
|
||||
|
||||
It("encodes an empty trace as an empty JSON array", func() {
|
||||
Expect(encodeSteps(nil, "")).To(Equal("[]"))
|
||||
Expect(DecodeTrace("[]", "")).To(BeEmpty())
|
||||
})
|
||||
|
||||
It("tolerates a row written before the column existed", func() {
|
||||
Expect(DecodeTrace("", "")).To(BeEmpty())
|
||||
})
|
||||
|
||||
// The hit detail repeats source_path byte for byte, and that column is on the same row.
|
||||
It("drops a hit detail that repeats sourcePath, and restores it on read", func() {
|
||||
path := "/music/artist/album/cover.jpg"
|
||||
steps := []TraceStep{{Candidate: "cover.*", Outcome: OutcomeHit, Detail: path}}
|
||||
encoded := encodeSteps(steps, path)
|
||||
Expect(encoded).NotTo(ContainSubstring(path))
|
||||
Expect(DecodeTrace(encoded, path)).To(Equal(steps))
|
||||
})
|
||||
|
||||
It("keeps a detail that differs from sourcePath", func() {
|
||||
steps := []TraceStep{{Candidate: "external:deezer", Outcome: OutcomeHit, Detail: "https://cdn/x.jpg"}}
|
||||
Expect(DecodeTrace(encodeSteps(steps, "/music/a/cover.jpg"), "/music/a/cover.jpg")).To(Equal(steps))
|
||||
})
|
||||
|
||||
// A row past ~1kB spills to an overflow page on these WITHOUT ROWID tables, which would
|
||||
// slow every scan; Detail is an error string on the failure paths, so it needs a bound.
|
||||
It("bounds a detail so one long error cannot inflate the row", func() {
|
||||
steps := []TraceStep{{Candidate: "decode", Outcome: OutcomeError, Detail: strings.Repeat("x", 5000)}}
|
||||
|
||||
got := DecodeTrace(encodeSteps(steps, ""), "")
|
||||
|
||||
Expect(len(got[0].Detail)).To(BeNumerically("<=", 210))
|
||||
Expect(got[0].Detail).To(HaveSuffix("..."))
|
||||
Expect(got[0].Candidate).To(Equal("decode"), "truncating the detail must not disturb the step")
|
||||
})
|
||||
|
||||
It("only restores sourcePath onto a detail-less hit", func() {
|
||||
steps := []TraceStep{{Candidate: "cover.*", Outcome: OutcomeMiss}}
|
||||
Expect(DecodeTrace(encodeSteps(steps, "/music/a/cover.jpg"), "/music/a/cover.jpg")).To(Equal(steps))
|
||||
})
|
||||
})
|
||||
|
||||
var _ = Describe("chainTrace", func() {
|
||||
It("returns nil when no trace is attached", func() {
|
||||
Expect(traceFrom(context.Background())).To(BeNil())
|
||||
@ -113,63 +163,41 @@ var _ = Describe("chainState tracing", func() {
|
||||
})
|
||||
})
|
||||
|
||||
var _ = Describe("external gate tracing", func() {
|
||||
hit := func() (io.ReadCloser, string, error) {
|
||||
return io.NopCloser(strings.NewReader("x")), "http://img", nil
|
||||
}
|
||||
miss := func() (io.ReadCloser, string, error) { return nil, "", agents.ErrNotFound }
|
||||
boom := func() (io.ReadCloser, string, error) { return nil, "", errors.New("returned status 429") }
|
||||
var _ = Describe("external agent tracing", func() {
|
||||
var (
|
||||
t *ChainTrace
|
||||
ctx context.Context
|
||||
body io.ReadCloser
|
||||
)
|
||||
BeforeEach(func() {
|
||||
t = &ChainTrace{}
|
||||
ctx = withTrace(context.Background(), t)
|
||||
body = io.NopCloser(strings.NewReader("x"))
|
||||
})
|
||||
|
||||
It("records a hit with the image path", func() {
|
||||
t := &ChainTrace{}
|
||||
g := tracingGate(t, passthroughGate)
|
||||
|
||||
r, _, err := g("deezer", hit)
|
||||
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(r).ToNot(BeNil())
|
||||
recordAgent(ctx, "deezer", body, "http://img", nil)
|
||||
Expect(t.Steps()).To(Equal([]TraceStep{
|
||||
{Candidate: "external:deezer", Outcome: OutcomeHit, Detail: "http://img"},
|
||||
}))
|
||||
})
|
||||
|
||||
It("records a miss for a not-found", func() {
|
||||
t := &ChainTrace{}
|
||||
_, _, _ = tracingGate(t, passthroughGate)("deezer", miss)
|
||||
recordAgent(ctx, "deezer", nil, "", agents.ErrNotFound)
|
||||
Expect(t.Steps()[0].Outcome).To(Equal(OutcomeMiss))
|
||||
})
|
||||
|
||||
It("records a miss for a model not-found", func() {
|
||||
t := &ChainTrace{}
|
||||
notFound := func() (io.ReadCloser, string, error) { return nil, "", model.ErrNotFound }
|
||||
_, _, _ = tracingGate(t, passthroughGate)("deezer", notFound)
|
||||
recordAgent(ctx, "deezer", nil, "", model.ErrNotFound)
|
||||
Expect(t.Steps()[0].Outcome).To(Equal(OutcomeMiss),
|
||||
"both not-found flavours are definitive answers, not faults")
|
||||
})
|
||||
|
||||
It("records an error with its reason", func() {
|
||||
t := &ChainTrace{}
|
||||
_, _, _ = tracingGate(t, passthroughGate)("apple-music", boom)
|
||||
recordAgent(ctx, "apple-music", nil, "", errors.New("returned status 429"))
|
||||
Expect(t.Steps()[0].Outcome).To(Equal(OutcomeError))
|
||||
Expect(t.Steps()[0].Detail).To(ContainSubstring("429"))
|
||||
})
|
||||
|
||||
It("never calls the agent in offline mode", func() {
|
||||
t := &ChainTrace{}
|
||||
called := false
|
||||
counting := func() (io.ReadCloser, string, error) {
|
||||
called = true
|
||||
return hit()
|
||||
}
|
||||
|
||||
_, _, err := offlineGate(t)("deezer", counting)
|
||||
|
||||
Expect(called).To(BeFalse(), "offline mode must not perform external requests")
|
||||
Expect(err).To(MatchError(errOfflineSkipped))
|
||||
Expect(t.Steps()).To(Equal([]TraceStep{
|
||||
{Candidate: "external:deezer", Outcome: OutcomeWouldTry},
|
||||
}))
|
||||
})
|
||||
})
|
||||
|
||||
var _ = Describe("resolveAlbum tracing", func() {
|
||||
@ -409,51 +437,51 @@ var _ = Describe("NewTracingResolver", func() {
|
||||
t = &ChainTrace{}
|
||||
})
|
||||
|
||||
Context("offline", func() {
|
||||
Context("resolving", func() {
|
||||
var fake *fakeImageAgent
|
||||
|
||||
BeforeEach(func() {
|
||||
fake = &fakeImageAgent{name: "offline-probe"}
|
||||
// Misses, so the chain falls through to the local tier and both are traced.
|
||||
fake = &fakeImageAgent{name: "probe", err: agents.ErrNotFound}
|
||||
albumRepo.SetData(model.Albums{{
|
||||
ID: "al1", Name: "Album", EmbedArtPath: "tests/fixtures/artist/an-album/test.mp3", FolderIDs: []string{"f1"},
|
||||
}})
|
||||
artistRepo.SetData(model.Artists{{ID: "ar1", Name: "Artist"}})
|
||||
})
|
||||
|
||||
It("reports the external tier without asking any agent", func() {
|
||||
source, err := NewTracingResolver(ds, imageAgents(fake), ffm, t, false).Resolve(context.Background(), model.KindAlbumArtwork, "al1")
|
||||
It("asks the agents and records what each answered", func() {
|
||||
source, err := NewTracingResolver(ds, imageAgents(fake), ffm, t, true).Resolve(context.Background(), model.KindAlbumArtwork, "al1")
|
||||
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(source).To(Equal("embedded"))
|
||||
Expect(fake.albumCalls).To(BeZero(), "offline mode must not add load to an external provider")
|
||||
Expect(t.Steps()).To(ContainElement(TraceStep{Candidate: "external:offline-probe", Outcome: OutcomeWouldTry}))
|
||||
Expect(fake.albumCalls).To(Equal(1))
|
||||
Expect(t.Steps()).To(ContainElement(TraceStep{Candidate: "external:probe", Outcome: OutcomeMiss}))
|
||||
})
|
||||
|
||||
It("records the local chain steps too", func() {
|
||||
_, err := NewTracingResolver(ds, imageAgents(fake), ffm, t, false).Resolve(context.Background(), model.KindAlbumArtwork, "al1")
|
||||
_, err := NewTracingResolver(ds, imageAgents(fake), ffm, t, true).Resolve(context.Background(), model.KindAlbumArtwork, "al1")
|
||||
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
last := t.Steps()[len(t.Steps())-1]
|
||||
Expect(last.Candidate).To(Equal("embedded"), "the local chain must be traced, not just the external gate")
|
||||
Expect(last.Candidate).To(Equal("embedded"), "the local chain must be traced, not just the external tier")
|
||||
Expect(last.Outcome).To(Equal(OutcomeHit))
|
||||
})
|
||||
|
||||
It("never persists artwork state", func() {
|
||||
_, err := NewTracingResolver(ds, imageAgents(fake), ffm, t, false).Resolve(context.Background(), model.KindAlbumArtwork, "al1")
|
||||
_, err := NewTracingResolver(ds, imageAgents(fake), ffm, t, true).Resolve(context.Background(), model.KindAlbumArtwork, "al1")
|
||||
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(artworkRepo.ItemData).To(BeEmpty(),
|
||||
"an offline resolution carries extError, which must never be recorded as a real provider failure")
|
||||
"explain is read-only; a diagnostic walk must never become the stored answer")
|
||||
Expect(queueRepo.Data).To(BeEmpty())
|
||||
})
|
||||
|
||||
It("resolves an artist without persisting anything", func() {
|
||||
source, err := NewTracingResolver(ds, imageAgents(fake), ffm, t, false).Resolve(context.Background(), model.KindArtistArtwork, "ar1")
|
||||
source, err := NewTracingResolver(ds, imageAgents(fake), ffm, t, true).Resolve(context.Background(), model.KindArtistArtwork, "ar1")
|
||||
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(source).To(BeEmpty())
|
||||
Expect(fake.artistCalls).To(BeZero())
|
||||
Expect(t.Steps()).To(ContainElement(TraceStep{Candidate: "external:offline-probe", Outcome: OutcomeWouldTry}))
|
||||
Expect(fake.artistCalls).To(Equal(1))
|
||||
Expect(artworkRepo.ItemData).To(BeEmpty())
|
||||
Expect(queueRepo.Data).To(BeEmpty())
|
||||
})
|
||||
@ -465,31 +493,39 @@ var _ = Describe("NewTracingResolver", func() {
|
||||
ID: "al2", Name: "Album", EmbedArtPath: "tests/fixtures/artist/an-album/no-such-file.mp3", FolderIDs: []string{"f1"},
|
||||
}})
|
||||
|
||||
source, err := NewTracingResolver(ds, imageAgents(fake), ffm, t, false).Resolve(context.Background(), model.KindAlbumArtwork, "al2")
|
||||
source, err := NewTracingResolver(ds, imageAgents(fake), ffm, t, true).Resolve(context.Background(), model.KindAlbumArtwork, "al2")
|
||||
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(source).To(Equal("embedded"))
|
||||
Expect(ffm.IsClosed()).To(BeTrue(), "nothing downstream closes it, so a leak is one file handle per invocation")
|
||||
})
|
||||
|
||||
// Serving falls back disc -> album and track -> disc -> album. The resolver does not, but
|
||||
// if it ever did, an explain without --live would start calling providers uninvited.
|
||||
It("cannot reach a provider without live, whatever the chain does", func() {
|
||||
conf.Server.DiscArtPriority = "external, cover.*"
|
||||
conf.Server.CoverArtPriority = "external, cover.*"
|
||||
conf.Server.EnableMediaFileCoverArt = true
|
||||
mfRepo := tests.CreateMockMediaFileRepo()
|
||||
mfRepo.SetData(model.MediaFiles{{ID: "mf1", LibraryID: 0, HasCoverArt: true,
|
||||
Path: "tests/fixtures/artist/an-album/test.mp3"}})
|
||||
ds.MockedMediaFile = mfRepo
|
||||
offline := NewTracingResolver(ds, imageAgents(fake), ffm, t, false)
|
||||
|
||||
_, err := offline.Resolve(context.Background(), model.KindDiscArtwork, "al1:1")
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
_, err = offline.Resolve(context.Background(), model.KindMediaFileArtwork, "mf1")
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
Expect(fake.albumCalls).To(BeZero())
|
||||
Expect(fake.artistCalls).To(BeZero())
|
||||
})
|
||||
|
||||
It("propagates a lookup error", func() {
|
||||
_, err := NewTracingResolver(ds, imageAgents(fake), ffm, t, false).Resolve(context.Background(), model.KindAlbumArtwork, "nope")
|
||||
_, err := NewTracingResolver(ds, imageAgents(fake), ffm, t, true).Resolve(context.Background(), model.KindAlbumArtwork, "nope")
|
||||
Expect(err).To(MatchError(model.ErrNotFound))
|
||||
})
|
||||
})
|
||||
|
||||
It("asks the agents when live is true", func() {
|
||||
fake := &fakeImageAgent{name: "live-probe", err: agents.ErrNotFound}
|
||||
albumRepo.SetData(model.Albums{{
|
||||
ID: "al1", Name: "Album", EmbedArtPath: "tests/fixtures/artist/an-album/test.mp3", FolderIDs: []string{"f1"},
|
||||
}})
|
||||
|
||||
_, err := NewTracingResolver(ds, imageAgents(fake), ffm, t, true).Resolve(context.Background(), model.KindAlbumArtwork, "al1")
|
||||
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(fake.albumCalls).To(Equal(1))
|
||||
Expect(t.Steps()).To(ContainElement(TraceStep{Candidate: "external:live-probe", Outcome: OutcomeMiss}))
|
||||
})
|
||||
})
|
||||
|
||||
var _ = Describe("resolveDisc tracing", func() {
|
||||
|
||||
@ -235,6 +235,8 @@ func (w *Worker) broadcastRefresh(ctx context.Context, found []model.ArtworkQueu
|
||||
|
||||
func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) (outcome, *acquired) {
|
||||
item.ImageType = cmp.Or(item.ImageType, model.ImageTypePrimary)
|
||||
trace := &ChainTrace{}
|
||||
ctx = withTrace(ctx, trace)
|
||||
out, got := w.proc.acquire(ctx, item)
|
||||
|
||||
queue := w.proc.ds.ArtworkQueue(ctx)
|
||||
@ -247,10 +249,11 @@ func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) (outc
|
||||
}
|
||||
case outcomeFoundStale, outcomeFailed:
|
||||
retryAt := time.Now().Add(backoff(item.Attempts))
|
||||
encoded := trace.encode("")
|
||||
if retryAt.Before(item.EnqueuedAt.Add(giveUpAfter)) {
|
||||
// A mid-flight re-enqueue reset retry_at; stale backoff must not stomp its
|
||||
// fresh, immediate eligibility.
|
||||
if err := queue.MarkFailedIfUnchanged(item.ItemKind, item.ItemID, item.ImageType, item.RetryAt, retryAt); err != nil {
|
||||
if err := queue.MarkFailedIfUnchanged(item.ItemKind, item.ItemID, item.ImageType, item.RetryAt, retryAt, encoded); err != nil {
|
||||
log.Warn(ctx, "Artwork: Could not reschedule failed queue item", "kind", item.ItemKind, "id", item.ItemID, err)
|
||||
}
|
||||
log.Debug(ctx, "Artwork: Rescheduled item", "kind", item.ItemKind, "id", item.ItemID,
|
||||
@ -265,6 +268,9 @@ func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) (outc
|
||||
writeAbsent(ctx, w.proc.ds.Artwork(ctx), item)
|
||||
settled = "recorded absent"
|
||||
}
|
||||
// The queue row is about to go, taking the only record of the failure with it. This write is
|
||||
// unconditional (not CAS-guarded) — safe only because the drain resolves each item serially.
|
||||
w.recordGiveUp(ctx, item, encoded)
|
||||
log.Info(ctx, "Artwork: Retry budget exhausted, giving up", "kind", item.ItemKind, "id", item.ItemID,
|
||||
"outcome", out, "attempts", item.Attempts+1, "budget", giveUpAfter, "settled", settled)
|
||||
if err := queue.DeleteIfUnchanged(item.ItemKind, item.ItemID, item.ImageType, item.RetryAt); err != nil {
|
||||
@ -274,6 +280,18 @@ func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) (outc
|
||||
return out, got
|
||||
}
|
||||
|
||||
// recordGiveUp keeps the last failure on the state row after the queue row is deleted. An item
|
||||
// that never resolved has no row to update, and creating one would settle it absent.
|
||||
func (w *Worker) recordGiveUp(ctx context.Context, item model.ArtworkQueueItem, trace string) {
|
||||
kind, ok := model.ParseKind(item.ItemKind)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if err := w.proc.ds.Artwork(ctx).PutLastFailure(kind, item.ItemID, item.ImageType, trace); err != nil {
|
||||
log.Warn(ctx, "Artwork: Could not record the last failure", "kind", item.ItemKind, "id", item.ItemID, err)
|
||||
}
|
||||
}
|
||||
|
||||
func (w *Worker) hasResolvedArtwork(ctx context.Context, item model.ArtworkQueueItem) bool {
|
||||
kind, ok := model.ParseKind(item.ItemKind)
|
||||
if !ok {
|
||||
|
||||
@ -95,6 +95,17 @@ func (f *fakeEventBroker) getEvents() []events.Event {
|
||||
|
||||
var _ events.Broker = (*fakeEventBroker)(nil)
|
||||
|
||||
// expireQueued ages a row past the retry budget, so the next drain settles it instead of retrying.
|
||||
func expireQueued(q *tests.MockArtworkQueueRepo, id string) {
|
||||
GinkgoHelper()
|
||||
for k, v := range q.Data {
|
||||
if v.ItemID == id {
|
||||
v.EnqueuedAt = time.Now().Add(-(giveUpAfter + time.Hour))
|
||||
q.Data[k] = v
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func findQueued(q *tests.MockArtworkQueueRepo, kind, id string) *model.ArtworkQueueItem {
|
||||
for _, it := range q.Data {
|
||||
if it.ItemKind == kind && it.ItemID == id {
|
||||
@ -318,12 +329,7 @@ var _ = Describe("Worker", func() {
|
||||
w = NewWorker(ds, store, ag, ffm, broker, imgCache)
|
||||
Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ItemKind: "al", ItemID: "al9"})).To(Succeed())
|
||||
// Age the row past the retry budget.
|
||||
for k, v := range queueRepo.Data {
|
||||
if v.ItemID == "al9" {
|
||||
v.EnqueuedAt = time.Now().Add(-(giveUpAfter + time.Hour))
|
||||
queueRepo.Data[k] = v
|
||||
}
|
||||
}
|
||||
expireQueued(queueRepo, "al9")
|
||||
|
||||
n, err := w.drain(ctx, 1)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
@ -345,12 +351,7 @@ var _ = Describe("Worker", func() {
|
||||
imageAgents(&fakeImageAgent{name: "failAgent", err: errors.New("agent timed out")})
|
||||
w = NewWorker(ds, store, ag, ffm, broker, imgCache)
|
||||
Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ItemKind: "al", ItemID: "al10"})).To(Succeed())
|
||||
for k, v := range queueRepo.Data {
|
||||
if v.ItemID == "al10" {
|
||||
v.EnqueuedAt = time.Now().Add(-(giveUpAfter + time.Hour))
|
||||
queueRepo.Data[k] = v
|
||||
}
|
||||
}
|
||||
expireQueued(queueRepo, "al10")
|
||||
|
||||
n, err := w.drain(ctx, 1)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
@ -362,6 +363,67 @@ var _ = Describe("Worker", func() {
|
||||
Expect(ia.Hash).To(Equal("cafebabe"), "a persistent outage must not discard served art")
|
||||
})
|
||||
|
||||
It("records on the queue row why the last attempt failed", func() {
|
||||
conf.Server.CoverArtPriority = "external"
|
||||
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{{ID: "al11", Name: "Album"}})
|
||||
imageAgents(&fakeImageAgent{name: "failAgent", err: errors.New("agent timed out")})
|
||||
w = NewWorker(ds, store, ag, ffm, broker, imgCache)
|
||||
Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ItemKind: "al", ItemID: "al11"})).To(Succeed())
|
||||
|
||||
_, err := w.drain(ctx, 1)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
it := findQueued(queueRepo, "al", "al11")
|
||||
Expect(it).ToNot(BeNil())
|
||||
Expect(DecodeTrace(it.Trace, "")).To(ContainElement(SatisfyAll(
|
||||
HaveField("Candidate", "external:failAgent"),
|
||||
HaveField("Outcome", OutcomeError),
|
||||
HaveField("Detail", ContainSubstring("agent timed out")),
|
||||
)), "a retrying row must say why it is retrying")
|
||||
})
|
||||
|
||||
// The give-up path settles absent before recording, so the row exists by the time the
|
||||
// failure is written. Recording first would silently lose it for every unresolved item.
|
||||
It("keeps the failure for an item that never resolved at all", func() {
|
||||
conf.Server.CoverArtPriority = "external"
|
||||
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{{ID: "al13", Name: "Album"}})
|
||||
imageAgents(&fakeImageAgent{name: "failAgent", err: errors.New("agent timed out")})
|
||||
w = NewWorker(ds, store, ag, ffm, broker, imgCache)
|
||||
Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ItemKind: "al", ItemID: "al13"})).To(Succeed())
|
||||
expireQueued(queueRepo, "al13")
|
||||
|
||||
_, err := w.drain(ctx, 1)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
ia, err := artRepo.GetItemArtwork(model.KindAlbumArtwork, "al13", model.ImageTypePrimary)
|
||||
Expect(err).ToNot(HaveOccurred(), "settling absent must create the row the failure is written to")
|
||||
Expect(ia.Hash).To(BeEmpty())
|
||||
Expect(DecodeTrace(ia.LastFailure, "")).ToNot(BeEmpty())
|
||||
})
|
||||
|
||||
It("keeps the failure on the state row after the queue row is deleted", func() {
|
||||
conf.Server.CoverArtPriority = "external"
|
||||
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{{ID: "al12", Name: "Album"}})
|
||||
Expect(artRepo.PutItemArtwork(&model.ItemArtwork{
|
||||
ItemKind: "al", ItemID: "al12", ImageType: model.ImageTypePrimary,
|
||||
Hash: "cafebabe", Source: "external:lastfm",
|
||||
})).To(Succeed())
|
||||
imageAgents(&fakeImageAgent{name: "failAgent", err: errors.New("agent timed out")})
|
||||
w = NewWorker(ds, store, ag, ffm, broker, imgCache)
|
||||
Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ItemKind: "al", ItemID: "al12"})).To(Succeed())
|
||||
expireQueued(queueRepo, "al12")
|
||||
|
||||
_, err := w.drain(ctx, 1)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
Expect(findQueued(queueRepo, "al", "al12")).To(BeNil())
|
||||
ia, err := artRepo.GetItemArtwork(model.KindAlbumArtwork, "al12", model.ImageTypePrimary)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(DecodeTrace(ia.LastFailure, "")).ToNot(BeEmpty(),
|
||||
"the queue row is gone, so this is the only remaining record of the failure")
|
||||
Expect(ia.Hash).To(Equal("cafebabe"), "recording the failure must not disturb the served art")
|
||||
})
|
||||
|
||||
// Media files are excluded from RecheckKinds, so an absent row here would never be
|
||||
// revisited: a transient read error would look permanent.
|
||||
It("does not settle absent on exhaustion for a kind with no recheck path", func() {
|
||||
@ -371,12 +433,7 @@ var _ = Describe("Worker", func() {
|
||||
{ID: "mfX", LibraryID: 0, Path: "tests/fixtures/artist/an-album/gone.mp3", HasCoverArt: true},
|
||||
})
|
||||
Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ItemKind: "mf", ItemID: "mfX"})).To(Succeed())
|
||||
for k, v := range queueRepo.Data {
|
||||
if v.ItemID == "mfX" {
|
||||
v.EnqueuedAt = time.Now().Add(-(giveUpAfter + time.Hour))
|
||||
queueRepo.Data[k] = v
|
||||
}
|
||||
}
|
||||
expireQueued(queueRepo, "mfX")
|
||||
|
||||
n, err := w.drain(ctx, 1)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
@ -386,6 +443,8 @@ var _ = Describe("Worker", func() {
|
||||
_, err = artRepo.GetItemArtwork(model.KindMediaFileArtwork, "mfX", model.ImageTypePrimary)
|
||||
Expect(err).To(MatchError(model.ErrNotFound),
|
||||
"no row leaves the track unresolved, so a later view can still recover it")
|
||||
// Known gap: with no row and no absent settle, there is nowhere to keep the failure.
|
||||
// Creating one here would write an empty hash, which every reader treats as absent.
|
||||
})
|
||||
|
||||
It("resolves a private playlist under an admin context instead of failing forever", func() {
|
||||
|
||||
@ -0,0 +1,9 @@
|
||||
-- +goose Up
|
||||
ALTER TABLE item_artwork ADD COLUMN trace jsonb NOT NULL DEFAULT '[]';
|
||||
ALTER TABLE item_artwork ADD COLUMN last_failure jsonb NOT NULL DEFAULT '[]';
|
||||
ALTER TABLE artwork_queue ADD COLUMN trace jsonb NOT NULL DEFAULT '[]';
|
||||
|
||||
-- +goose Down
|
||||
ALTER TABLE artwork_queue DROP COLUMN trace;
|
||||
ALTER TABLE item_artwork DROP COLUMN last_failure;
|
||||
ALTER TABLE item_artwork DROP COLUMN trace;
|
||||
@ -51,6 +51,10 @@ type ItemArtwork struct {
|
||||
SourcePath string `structs:"source_path"`
|
||||
// RefMtime is SourcePath's mtime (unix-nanoseconds) at resolution; 0 when there is no SourcePath.
|
||||
RefMtime int64 `structs:"ref_mtime"`
|
||||
// Trace is the encoded walk that produced this state; LastFailure is the walk of the attempt
|
||||
// that exhausted the retry budget. Both are JSON, read back with artwork.DecodeTrace.
|
||||
Trace string `structs:"trace"`
|
||||
LastFailure string `structs:"last_failure"`
|
||||
// Nullable in the schema, but every insert must set them: these non-pointer fields cannot scan NULL.
|
||||
AttemptedAt time.Time `structs:"attempted_at"`
|
||||
UpdatedAt time.Time `structs:"updated_at"`
|
||||
@ -91,6 +95,8 @@ type ArtworkQueueItem struct {
|
||||
Attempts int `structs:"attempts"`
|
||||
RetryAt time.Time `structs:"retry_at"`
|
||||
EnqueuedAt time.Time `structs:"enqueued_at"`
|
||||
// Trace is why the last attempt failed. Only Get reads it; the drain projects it away.
|
||||
Trace string `structs:"trace"`
|
||||
}
|
||||
|
||||
// Queue priorities: higher drains first.
|
||||
@ -109,6 +115,8 @@ type ArtworkRepository interface {
|
||||
PurgeOrphans(createdBefore time.Time) (int64, error)
|
||||
GetItemArtwork(kind Kind, id, imageType string) (*ItemArtwork, error)
|
||||
PutItemArtwork(ia *ItemArtwork) error
|
||||
// PutLastFailure records the trace of the attempt that exhausted the retry budget.
|
||||
PutLastFailure(kind Kind, id, imageType, trace string) error
|
||||
DeleteForItems(kind Kind, ids []string) error
|
||||
// GetInfoForItems hydrates a page in one batched query.
|
||||
GetInfoForItems(kind Kind, ids []string) (map[string]ItemArtworkInfo, error)
|
||||
@ -145,7 +153,7 @@ type ArtworkQueueRepository interface {
|
||||
DequeueBatch(n int, kinds ...string) ([]ArtworkQueueItem, error)
|
||||
// MarkFailedIfUnchanged applies the failure backoff only while retry_at still matches
|
||||
// seenRetryAt, so a concurrent re-enqueue keeps its fresh eligibility.
|
||||
MarkFailedIfUnchanged(kind, id, imageType string, seenRetryAt, retryAt time.Time) error
|
||||
MarkFailedIfUnchanged(kind, id, imageType string, seenRetryAt, retryAt time.Time, trace string) error
|
||||
// DeleteIfUnchanged deletes only while retry_at still matches, sparing a concurrent re-enqueue.
|
||||
DeleteIfUnchanged(kind, id, imageType string, retryAt time.Time) error
|
||||
Count() (int64, error)
|
||||
|
||||
@ -18,6 +18,7 @@ import (
|
||||
const enqueueChunkSize = 100
|
||||
|
||||
// Every insert writes these, in this order; the INSERT..SELECT forms must project them to match.
|
||||
// DequeueBatch also selects exactly these, to leave the drain's rows free of the trace it never reads.
|
||||
var enqueueColumns = []string{"item_kind", "item_id", "image_type", "priority", "attempts", "retry_at", "enqueued_at"}
|
||||
|
||||
type artworkQueueRepository struct {
|
||||
@ -42,11 +43,12 @@ func (r *artworkQueueRepository) Get(kind model.Kind, id, imageType string) (*mo
|
||||
return &res, nil
|
||||
}
|
||||
|
||||
// Enqueue also resets enqueued_at, so a fresh request does not inherit an old row's spent retry budget.
|
||||
// Enqueue starts a fresh lifecycle: it resets enqueued_at (so a fresh request does not inherit an old
|
||||
// row's spent retry budget) and clears trace (so explain does not show a prior failure at attempts 0).
|
||||
func (r *artworkQueueRepository) Enqueue(items ...model.ArtworkQueueItem) error {
|
||||
return r.enqueue(`ON CONFLICT (item_kind, item_id, image_type) DO UPDATE SET
|
||||
priority = MAX(priority, excluded.priority), retry_at = excluded.retry_at,
|
||||
attempts = 0, enqueued_at = excluded.enqueued_at`, items)
|
||||
attempts = 0, enqueued_at = excluded.enqueued_at, trace = '[]'`, items)
|
||||
}
|
||||
|
||||
func (r *artworkQueueRepository) EnqueuePreservingBackoff(items ...model.ArtworkQueueItem) error {
|
||||
@ -159,7 +161,7 @@ func (r *artworkQueueRepository) enqueue(conflict string, items []model.ArtworkQ
|
||||
}
|
||||
|
||||
func (r *artworkQueueRepository) DequeueBatch(n int, kinds ...string) ([]model.ArtworkQueueItem, error) {
|
||||
sel := Select("*").From(r.tableName).
|
||||
sel := Select(enqueueColumns...).From(r.tableName).
|
||||
Where(LtOrEq{"retry_at": time.Now()}).
|
||||
OrderBy("priority DESC", "enqueued_at ASC").
|
||||
Limit(uint64(n))
|
||||
@ -171,10 +173,11 @@ func (r *artworkQueueRepository) DequeueBatch(n int, kinds ...string) ([]model.A
|
||||
return res, err
|
||||
}
|
||||
|
||||
func (r *artworkQueueRepository) MarkFailedIfUnchanged(kind, id, imageType string, seenRetryAt, retryAt time.Time) error {
|
||||
func (r *artworkQueueRepository) MarkFailedIfUnchanged(kind, id, imageType string, seenRetryAt, retryAt time.Time, trace string) error {
|
||||
upd := Update(r.tableName).
|
||||
Set("attempts", Expr("attempts + 1")).
|
||||
Set("retry_at", retryAt).
|
||||
Set("trace", trace).
|
||||
Where(Eq{"item_kind": kind, "item_id": id, "image_type": imageType, "retry_at": seenRetryAt})
|
||||
_, err := r.executeSQL(upd)
|
||||
return err
|
||||
|
||||
@ -127,19 +127,42 @@ var _ = Describe("ArtworkQueueRepository", func() {
|
||||
Expect(repo.Enqueue(item("al", "m1", model.ArtworkPriorityScan))).To(Succeed())
|
||||
|
||||
future := time.Now().Add(48 * time.Hour)
|
||||
Expect(repo.MarkFailedIfUnchanged("al", "m1", model.ImageTypePrimary, original, future)).To(Succeed())
|
||||
Expect(repo.MarkFailedIfUnchanged("al", "m1", model.ImageTypePrimary, original, future, "[]")).To(Succeed())
|
||||
got, _ = repo.DequeueBatch(10)
|
||||
Expect(got).To(HaveLen(1), "the fresh re-enqueue stays immediately eligible")
|
||||
Expect(got[0].Attempts).To(BeZero(), "re-enqueue clears attempts, and the stale failure must not bump them")
|
||||
current := got[0].RetryAt
|
||||
|
||||
Expect(repo.MarkFailedIfUnchanged("al", "m1", model.ImageTypePrimary, current, future)).To(Succeed())
|
||||
Expect(repo.MarkFailedIfUnchanged("al", "m1", model.ImageTypePrimary, current, future, `[{"c":"read","o":"error"}]`)).To(Succeed())
|
||||
got, _ = repo.DequeueBatch(10)
|
||||
Expect(got).To(BeEmpty(), "backed-off row is hidden until the future retry_at")
|
||||
all, _ := repo.Count()
|
||||
Expect(all).To(Equal(int64(1)))
|
||||
})
|
||||
|
||||
It("Enqueue clears a prior lifecycle's failure trace; EnqueuePreservingBackoff keeps it", func() {
|
||||
// Fail an attempt so the queue row carries a failure trace.
|
||||
Expect(repo.Enqueue(item("al", "t1", model.ArtworkPriorityScan))).To(Succeed())
|
||||
backOff("al", "t1", time.Now().Add(-time.Hour))
|
||||
got, _ := repo.DequeueBatch(10)
|
||||
Expect(got).To(HaveLen(1))
|
||||
future := time.Now().Add(48 * time.Hour)
|
||||
Expect(repo.MarkFailedIfUnchanged("al", "t1", model.ImageTypePrimary, got[0].RetryAt, future, `[{"c":"read","o":"error"}]`)).To(Succeed())
|
||||
|
||||
// A continuation of the same lifecycle must retain the trace.
|
||||
Expect(repo.EnqueuePreservingBackoff(item("al", "t1", model.ArtworkPriorityBump))).To(Succeed())
|
||||
kept, err := repo.Get(model.KindAlbumArtwork, "t1", model.ImageTypePrimary)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(kept.Trace).To(Equal(`[{"c":"read","o":"error"}]`))
|
||||
|
||||
// A fresh Enqueue resets attempts to 0, so the stale failure trace must be cleared with it.
|
||||
Expect(repo.Enqueue(item("al", "t1", model.ArtworkPriorityScan))).To(Succeed())
|
||||
fresh, err := repo.Get(model.KindAlbumArtwork, "t1", model.ImageTypePrimary)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(fresh.Attempts).To(BeZero())
|
||||
Expect(fresh.Trace).To(Equal("[]"), "a fresh lifecycle has no last-attempt trace")
|
||||
})
|
||||
|
||||
It("Enqueue restarts the retry budget an existing row had spent", func() {
|
||||
Expect(repo.Enqueue(item("al", "e1", model.ArtworkPriorityScan))).To(Succeed())
|
||||
backOff("al", "e1", time.Now().Add(-time.Hour))
|
||||
|
||||
@ -134,11 +134,21 @@ func (r *artworkRepository) PutItemArtwork(ia *model.ItemArtwork) error {
|
||||
}
|
||||
ins := Insert(itemArtworkTable).SetMap(values).Suffix(`ON CONFLICT (item_kind, item_id, image_type) DO UPDATE SET
|
||||
hash=excluded.hash, source=excluded.source, source_path=excluded.source_path, ref_mtime=excluded.ref_mtime,
|
||||
trace=excluded.trace, last_failure=excluded.last_failure,
|
||||
attempted_at=excluded.attempted_at, updated_at=excluded.updated_at`)
|
||||
_, err = r.items.executeSQL(ins)
|
||||
return err
|
||||
}
|
||||
|
||||
// PutLastFailure records why an item exhausted its retry budget. It only updates an existing row:
|
||||
// inserting one would write an empty hash, which the rest of the system reads as a settled absent.
|
||||
func (r *artworkRepository) PutLastFailure(kind model.Kind, id, imageType, trace string) error {
|
||||
upd := Update(itemArtworkTable).Set("last_failure", trace).
|
||||
Where(Eq{"item_kind": kind.Prefix(), "item_id": id, "image_type": imageType})
|
||||
_, err := r.items.executeSQL(upd)
|
||||
return err
|
||||
}
|
||||
|
||||
func (r *artworkRepository) DeleteForItems(kind model.Kind, ids []string) error {
|
||||
for chunk := range slices.Chunk(ids, artworkBatchSize) {
|
||||
if err := r.items.delete(Eq{"item_kind": kind.Prefix(), "item_id": chunk}); err != nil {
|
||||
|
||||
@ -28,6 +28,51 @@ var _ = Describe("ArtworkRepository", func() {
|
||||
repo = NewArtworkRepository(context.Background(), GetDBXBuilder())
|
||||
})
|
||||
|
||||
Context("resolution traces", func() {
|
||||
const traceJSON = `[{"c":"cover.*","o":"hit"}]`
|
||||
|
||||
It("round-trips the trace with the state row", func() {
|
||||
Expect(repo.PutItemArtwork(&model.ItemArtwork{ItemKind: "al", ItemID: "t1",
|
||||
ImageType: model.ImageTypePrimary, Hash: "h1", Trace: traceJSON})).To(Succeed())
|
||||
|
||||
got, err := repo.GetItemArtwork(model.KindAlbumArtwork, "t1", model.ImageTypePrimary)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(got.Trace).To(Equal(traceJSON))
|
||||
Expect(got.LastFailure).To(BeEmpty())
|
||||
})
|
||||
|
||||
It("replaces the trace when the item is resolved again", func() {
|
||||
Expect(repo.PutItemArtwork(&model.ItemArtwork{ItemKind: "al", ItemID: "t2",
|
||||
ImageType: model.ImageTypePrimary, Trace: traceJSON})).To(Succeed())
|
||||
Expect(repo.PutItemArtwork(&model.ItemArtwork{ItemKind: "al", ItemID: "t2",
|
||||
ImageType: model.ImageTypePrimary, Trace: `[{"c":"embedded","o":"hit"}]`})).To(Succeed())
|
||||
|
||||
got, _ := repo.GetItemArtwork(model.KindAlbumArtwork, "t2", model.ImageTypePrimary)
|
||||
Expect(got.Trace).To(Equal(`[{"c":"embedded","o":"hit"}]`))
|
||||
})
|
||||
|
||||
It("records a last failure on an existing row", func() {
|
||||
Expect(repo.PutItemArtwork(&model.ItemArtwork{ItemKind: "al", ItemID: "t3",
|
||||
ImageType: model.ImageTypePrimary, Hash: "h3"})).To(Succeed())
|
||||
|
||||
Expect(repo.PutLastFailure(model.KindAlbumArtwork, "t3", model.ImageTypePrimary,
|
||||
`[{"c":"decode","o":"error"}]`)).To(Succeed())
|
||||
|
||||
got, _ := repo.GetItemArtwork(model.KindAlbumArtwork, "t3", model.ImageTypePrimary)
|
||||
Expect(got.LastFailure).To(Equal(`[{"c":"decode","o":"error"}]`))
|
||||
Expect(got.Hash).To(Equal("h3"), "recording a failure must not disturb the served artwork")
|
||||
})
|
||||
|
||||
// Inserting here would write hash='', which every reader treats as a settled absent.
|
||||
It("never creates a row for an item that has no state", func() {
|
||||
Expect(repo.PutLastFailure(model.KindAlbumArtwork, "ghost", model.ImageTypePrimary,
|
||||
`[{"c":"decode","o":"error"}]`)).To(Succeed())
|
||||
|
||||
_, err := repo.GetItemArtwork(model.KindAlbumArtwork, "ghost", model.ImageTypePrimary)
|
||||
Expect(err).To(MatchError(model.ErrNotFound))
|
||||
})
|
||||
})
|
||||
|
||||
Context("image identity", func() {
|
||||
It("stores and retrieves an artwork by hash", func() {
|
||||
a := &model.Artwork{Hash: "abc123", Mime: "image/jpeg", Width: 500, Height: 500, SizeBytes: 1234, BlurHash: "LKO2?U%2Tw=w"}
|
||||
|
||||
@ -118,7 +118,7 @@ func (m *MockArtworkQueueRepo) DequeueBatch(n int, kinds ...string) ([]model.Art
|
||||
return res, nil
|
||||
}
|
||||
|
||||
func (m *MockArtworkQueueRepo) MarkFailedIfUnchanged(kind, id, imageType string, seenRetryAt, retryAt time.Time) error {
|
||||
func (m *MockArtworkQueueRepo) MarkFailedIfUnchanged(kind, id, imageType string, seenRetryAt, retryAt time.Time, trace string) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if m.Err != nil {
|
||||
@ -128,6 +128,7 @@ func (m *MockArtworkQueueRepo) MarkFailedIfUnchanged(kind, id, imageType string,
|
||||
if it, ok := m.Data[k]; ok && it.RetryAt.Equal(seenRetryAt) {
|
||||
it.Attempts++
|
||||
it.RetryAt = retryAt
|
||||
it.Trace = trace
|
||||
m.Data[k] = it
|
||||
}
|
||||
return nil
|
||||
|
||||
@ -122,6 +122,20 @@ func (m *MockArtworkRepo) GetItemArtwork(kind model.Kind, id, imageType string)
|
||||
return nil, model.ErrNotFound
|
||||
}
|
||||
|
||||
func (m *MockArtworkRepo) PutLastFailure(kind model.Kind, id, imageType, trace string) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if m.Err != nil {
|
||||
return m.Err
|
||||
}
|
||||
key := iaKey(kind.Prefix(), id, imageType)
|
||||
if ia, ok := m.ItemData[key]; ok {
|
||||
ia.LastFailure = trace
|
||||
m.ItemData[key] = ia
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *MockArtworkRepo) PutItemArtwork(ia *model.ItemArtwork) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user