fix(ebooks): account canceled and vanished claims accurately

A claim whose enrichment surfaced context.Canceled while the sweep was
still live was released uncounted, leaving it immediately reclaimable and
invisible to the no-progress circuit breaker. Let it flow through the
failure path as a transient error so it backs off and is counted; genuine
sweep shutdown still releases via the existing ctx check.

Claims discarded because the item vanished are terminal, not retried, so
report them in a new discarded counter instead of inflating deferred, and
surface the count in task progress output.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
rxwatcher
2026-07-22 13:43:57 +02:00
co-authored by Claude Fable 5
parent fcd7eb697d
commit d6cdf5c9e4
3 changed files with 47 additions and 8 deletions
+3 -5
View File
@@ -116,6 +116,7 @@ type EnrichmentRunResult struct {
NoMatch int `json:"no_match"`
Failed int `json:"failed"`
Deferred int `json:"deferred"`
Discarded int `json:"discarded"`
Remaining int `json:"remaining"`
HasMore bool `json:"-"`
}
@@ -282,6 +283,7 @@ func (e *Enricher) RunLimited(ctx context.Context, scope EnrichmentScope, maxCla
"no_match", result.NoMatch,
"failed", result.Failed,
"deferred", result.Deferred,
"discarded", result.Discarded,
"has_more", result.HasMore,
)
return result, runErr
@@ -325,7 +327,7 @@ func (e *Enricher) runQueueBatch(
if err := e.discardJob(queue, job); err != nil && !errors.Is(err, ErrEnrichmentLeaseLost) {
transitionErrs = append(transitionErrs, fmt.Errorf("%s: %w", job.ContentID, err))
} else if err == nil {
result.Deferred++
result.Discarded++
}
}
}
@@ -405,10 +407,6 @@ func (e *Enricher) runQueueBatch(
}
continue
}
if errors.Is(enrichErr, context.Canceled) {
recordTransitionError(item.ContentID, e.releaseJob(queue, job))
continue
}
if enrichErr != nil {
errorClass, retryAfter := classifyEnrichmentError(enrichErr)
transitionErr := e.failJob(
+39 -1
View File
@@ -402,7 +402,8 @@ func TestEnricherRunDiscardsClaimedRowsThatAreNoLongerEligible(t *testing.T) {
},
}
if _, err := e.Run(context.Background(), EnrichmentScopeIncremental); err != nil {
result, err := e.Run(context.Background(), EnrichmentScopeIncremental)
if err != nil {
t.Fatalf("Run() error = %v", err)
}
if got := strings.Join(queue.discarded, ","); got != "became-manga" {
@@ -414,6 +415,43 @@ func TestEnricherRunDiscardsClaimedRowsThatAreNoLongerEligible(t *testing.T) {
if len(queue.released) != 0 {
t.Fatalf("ineligible rows were released back into immediate churn: %v", queue.released)
}
if result.Discarded != 1 {
t.Fatalf("Discarded = %d, want 1", result.Discarded)
}
if result.Deferred != 1 {
t.Fatalf("Deferred = %d, want 1 (the skipped completion only); a discarded row is terminal, not deferred for retry", result.Deferred)
}
}
func TestEnricherRunCountsUpstreamCancellationAsTransientFailure(t *testing.T) {
queue := &fakeEnrichmentQueue{
jobs: []EnrichmentJob{{ContentID: "cancel-happy", Token: "cancel-token", Attempts: 1}},
}
e := &Enricher{
queue: queue,
batchSize: 1,
workers: 1,
loadClaimedItemsFn: func(context.Context, []EnrichmentJob) ([]enrichmentItemRow, error) {
return []enrichmentItemRow{{ContentID: "cancel-happy"}}, nil
},
enrichClaimedItemFn: func(context.Context, enrichmentItemRow) (EnrichmentOutcome, error) {
return "", context.Canceled
},
}
result, err := e.Run(context.Background(), EnrichmentScopeIncremental)
if err != nil {
t.Fatalf("Run() error = %v", err)
}
if got := queue.failed["cancel-happy"]; got != EnrichmentErrorTransient {
t.Fatalf("failure class = %q, want transient", got)
}
if result.Failed != 1 {
t.Fatalf("Failed = %d, want 1; an uncounted claim blinds the no-progress circuit breaker", result.Failed)
}
if len(queue.released) != 0 {
t.Fatalf("provider-canceled work was released into immediate reclaim churn: %v", queue.released)
}
}
func TestEnricherRunBoundsEachItemBelowTheLease(t *testing.T) {
@@ -231,6 +231,7 @@ func addEbookEnrichmentResult(total *ebooks.EnrichmentRunResult, batch ebooks.En
total.NoMatch += batch.NoMatch
total.Failed += batch.Failed
total.Deferred += batch.Deferred
total.Discarded += batch.Discarded
}
func reportEbookEnrichmentProgress(
@@ -246,12 +247,13 @@ func reportEbookEnrichmentProgress(
percent = 100
}
progress.Report(percent, fmt.Sprintf(
"Claimed %d, enriched %d, no match %d, failed %d, deferred %d, remaining %d",
"Claimed %d, enriched %d, no match %d, failed %d, deferred %d, discarded %d, remaining %d",
result.Claimed,
result.Enriched,
result.NoMatch,
result.Failed,
result.Deferred,
result.Discarded,
result.Remaining,
))
}
@@ -287,13 +289,14 @@ func reportEbookEnrichmentPause(
percent = 99
}
progress.Report(percent, fmt.Sprintf(
"%s Claimed %d, enriched %d, no match %d, failed %d, deferred %d, remaining %d",
"%s Claimed %d, enriched %d, no match %d, failed %d, deferred %d, discarded %d, remaining %d",
message,
result.Claimed,
result.Enriched,
result.NoMatch,
result.Failed,
result.Deferred,
result.Discarded,
result.Remaining,
))
}