fix(markers): stop retrying contribution conflicts (#559)

* fix(markers): stop retrying contribution conflicts

* fix(markers): claim contributions atomically

* fix(markers): recover stale contribution claims

internal/markers/contribute.go

@@ -2,6 +2,7 @@ package markers
 
 import (
 	"context"
+	"errors"
 	"log/slog"
 	"strconv"
 	"strings"
@@ -13,9 +14,14 @@ import (
 // OutcomeStatus values for a contribution attempt, in addition to the provider
 // SubmissionStatus* values.
 const (
+	OutcomeStatusConflict    = "conflict"
 	OutcomeStatusError       = "error"
 	OutcomeStatusRateLimited = "rate_limited"
 	OutcomeStatusSkipped     = "skipped"
+	contributionStatusClaim  = "submitting"
+	contributionSubmitLimit  = 2 * time.Minute
+	contributionClaimLease   = 15 * time.Minute
+	contributionRecordLimit  = 5 * time.Second
 )
 
 // ContributeOptions scopes a contribution run.
@@ -34,7 +40,7 @@ type ContributeOptions struct {
 type ContributionOutcome struct {
 	Provider     string
 	Segment      MarkerKind
-	Status       string // pending | accepted | rejected | error | rate_limited | skipped
+	Status       string // pending | accepted | rejected | conflict | error | rate_limited | skipped
 	SubmissionID string
 	Reason       string // skip reason or error message
 	RetryAfter   time.Duration
@@ -49,7 +55,7 @@ type providerConfigReader interface {
 // contributionRecorder is the audit surface ContributionService needs
 // (satisfied by *ContributionStore).
 type contributionRecorder interface {
-	AlreadySubmitted(ctx context.Context, fileID int, provider, segmentKind, contentHash string) (bool, error)
+	Claim(ctx context.Context, row ContributionRow, staleAfter time.Duration) (ContributionClaim, bool, error)
 	Record(ctx context.Context, row ContributionRow) error
 }
 
@@ -201,13 +207,26 @@ func (s *ContributionService) contributeSegment(
 	durMs := int64(file.Duration) * 1000
 	hash := ContentHash(seg.name, &startMs, &endMs, &durMs, contributionTargetParts(ids)...)
 
-	already, err := s.store.AlreadySubmitted(ctx, file.ID, providerID, seg.name, hash)
+	row := ContributionRow{
+		MediaFileID:      file.ID,
+		Provider:         providerID,
+		SegmentKind:      seg.name,
+		Source:           source,
+		SubmittedStartMs: &startMs,
+		SubmittedEndMs:   &endMs,
+		VideoDurationMs:  &durMs,
+		ContentHash:      hash,
+		Status:           contributionStatusClaim,
+	}
+	claim, claimed, err := s.store.Claim(ctx, row, contributionClaimLease)
 	if err != nil {
 		return ContributionOutcome{Provider: providerID, Segment: seg.kind, Status: OutcomeStatusError, Reason: err.Error()}, true
 	}
-	if already {
+	if !claimed {
 		return ContributionOutcome{Provider: providerID, Segment: seg.kind, Status: OutcomeStatusSkipped, Reason: "already submitted"}, true
 	}
+	row.ID = claim.ID
+	row.ClaimToken = claim.Token
 
 	startDur := time.Duration(*seg.start * float64(time.Second))
 	endDur := time.Duration(*seg.end * float64(time.Second))
@@ -222,24 +241,25 @@ func (s *ContributionService) contributeSegment(
 		Duration:      time.Duration(file.Duration) * time.Second,
 	}
 
-	row := ContributionRow{
-		MediaFileID:      file.ID,
-		Provider:         providerID,
-		SegmentKind:      seg.name,
-		Source:           source,
-		SubmittedStartMs: &startMs,
-		SubmittedEndMs:   &endMs,
-		VideoDurationMs:  &durMs,
-		ContentHash:      hash,
-	}
-
-	result, err := sub.SubmitMarker(ctx, req)
+	submitCtx, cancel := context.WithTimeout(ctx, contributionSubmitLimit)
+	result, err := sub.SubmitMarker(submitCtx, req)
+	cancel()
 	if err != nil {
 		msg := err.Error()
-		row.Status = OutcomeStatusError
 		row.Error = &msg
-		if recErr := s.store.Record(ctx, row); recErr != nil {
-			s.logger.WarnContext(ctx, "record contribution error failed", "file_id", file.ID, "provider", providerID, "segment", seg.name, "error", recErr)
+		var conflict *SubmissionConflictError
+		if errors.As(err, &conflict) && conflict != nil {
+			row.Status = OutcomeStatusConflict
+			if conflict.HTTPStatus > 0 {
+				status := conflict.HTTPStatus
+				row.HTTPStatus = &status
+			}
+		} else {
+			row.Status = OutcomeStatusError
+		}
+		s.recordContribution(ctx, row)
+		if row.Status == OutcomeStatusConflict {
+			return ContributionOutcome{Provider: providerID, Segment: seg.kind, Status: OutcomeStatusConflict, Reason: msg}, true
 		}
 		if after, ok := RetryAfter(err); ok {
 			return ContributionOutcome{Provider: providerID, Segment: seg.kind, Status: OutcomeStatusRateLimited, Reason: msg, RetryAfter: after}, true
@@ -255,12 +275,18 @@ func (s *ContributionService) contributeSegment(
 		id := result.ID
 		row.SubmissionID = &id
 	}
-	if err := s.store.Record(ctx, row); err != nil {
-		s.logger.WarnContext(ctx, "record contribution failed", "file_id", file.ID, "provider", providerID, "segment", seg.name, "error", err)
-	}
+	s.recordContribution(ctx, row)
 	return ContributionOutcome{Provider: providerID, Segment: seg.kind, Status: row.Status, SubmissionID: result.ID}, true
 }
 
+func (s *ContributionService) recordContribution(ctx context.Context, row ContributionRow) {
+	recordCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), contributionRecordLimit)
+	defer cancel()
+	if err := s.store.Record(recordCtx, row); err != nil {
+		s.logger.WarnContext(recordCtx, "record contribution failed", "file_id", row.MediaFileID, "provider", row.Provider, "segment", row.SegmentKind, "status", row.Status, "error", err)
+	}
+}
+
 func contributionTargetParts(ids ExternalIDs) []string {
 	return []string{
 		itemTypeName(ids.Kind),

internal/markers/contribute_test.go

@@ -2,31 +2,62 @@ package markers
 
 import (
 	"context"
+	"errors"
+	"fmt"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/Silo-Server/silo-server/internal/models"
+	"github.com/jackc/pgx/v5/pgconn"
 )
 
 type fakeSubmitter struct {
+	mu        sync.Mutex
 	id        string
 	submitted []SubmissionRequest
 	result    SubmissionResult
 	err       error
 	required  []string
+	started   chan struct{}
+	release   chan struct{}
+	onSubmit  func()
+	deadline  chan time.Time
+	startOnce sync.Once
 }
 
 func (f *fakeSubmitter) ID() string                                            { return f.id }
 func (f *fakeSubmitter) FetchMarkers(context.Context, Request) (Result, error) { return Result{}, nil }
-func (f *fakeSubmitter) SubmitMarker(_ context.Context, req SubmissionRequest) (SubmissionResult, error) {
+func (f *fakeSubmitter) SubmitMarker(ctx context.Context, req SubmissionRequest) (SubmissionResult, error) {
+	f.mu.Lock()
 	f.submitted = append(f.submitted, req)
-	if f.err != nil {
-		return SubmissionResult{}, f.err
+	err := f.err
+	result := f.result
+	started := f.started
+	release := f.release
+	onSubmit := f.onSubmit
+	deadline := f.deadline
+	f.mu.Unlock()
+	if deadline != nil {
+		observed, _ := ctx.Deadline()
+		deadline <- observed
 	}
-	if f.result.Status == "" {
+	if started != nil {
+		f.startOnce.Do(func() { close(started) })
+	}
+	if release != nil {
+		<-release
+	}
+	if onSubmit != nil {
+		onSubmit()
+	}
+	if err != nil {
+		return SubmissionResult{}, err
+	}
+	if result.Status == "" {
 		return SubmissionResult{ID: "id1", Status: SubmissionStatusPending}, nil
 	}
-	return f.result, nil
+	return result, nil
 }
 func (f *fakeSubmitter) FetchUserStats(context.Context) (UserStats, error) { return UserStats{}, nil }
 func (f *fakeSubmitter) SubmissionRequirements() SubmissionRequirements {
@@ -44,15 +75,42 @@ type fakeConfig map[string]ProviderConfig
 func (f fakeConfig) Get(p string) (ProviderConfig, bool) { c, ok := f[p]; return c, ok }
 
 type fakeRecorder struct {
+	mu       sync.Mutex
 	already  bool
 	recorded []ContributionRow
+	claims   map[string]string
+	next     int
 }
 
-func (f *fakeRecorder) AlreadySubmitted(context.Context, int, string, string, string) (bool, error) {
-	return f.already, nil
+func (f *fakeRecorder) Claim(_ context.Context, row ContributionRow, _ time.Duration) (ContributionClaim, bool, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	key := row.Provider + "|" + row.SegmentKind + "|" + row.ContentHash
+	if _, exists := f.claims[key]; f.already || exists {
+		return ContributionClaim{}, false, nil
+	}
+	if f.claims == nil {
+		f.claims = make(map[string]string)
+	}
+	f.next++
+	token := fmt.Sprintf("claim-%d", f.next)
+	f.claims[key] = token
+	return ContributionClaim{ID: key, Token: token}, true, nil
 }
-func (f *fakeRecorder) Record(_ context.Context, row ContributionRow) error {
+func (f *fakeRecorder) Record(ctx context.Context, row ContributionRow) error {
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	key := row.Provider + "|" + row.SegmentKind + "|" + row.ContentHash
+	if f.claims[key] != row.ClaimToken {
+		return errors.New("claim not found")
+	}
 	f.recorded = append(f.recorded, row)
+	if row.Status == OutcomeStatusError {
+		delete(f.claims, key)
+	}
 	return nil
 }
 
@@ -156,6 +214,190 @@ func TestContributeSkipsDuplicate(t *testing.T) {
 	}
 }
 
+func TestContributeSettlesConflictAndSkipsRetry(t *testing.T) {
+	sub := &fakeSubmitter{
+		id: "introdb",
+		err: &SubmissionConflictError{
+			Provider:   "introdb",
+			HTTPStatus: 409,
+			Message:    "already submitted",
+		},
+	}
+	file := newContribFile()
+	file.IntroStart, file.IntroEnd = floatPtr(0), floatPtr(60)
+	file.IntroMarkersSource = strPtr(models.MarkerSourceManual)
+	rec := &fakeRecorder{}
+	svc := newContribService(sub, fakeConfig{"introdb": {Provider: "introdb", ContributeEnabled: true}}, rec)
+
+	first, err := svc.ContributeFile(context.Background(), file, ContributeOptions{})
+	if err != nil {
+		t.Fatalf("first ContributeFile: %v", err)
+	}
+	second, err := svc.ContributeFile(context.Background(), file, ContributeOptions{})
+	if err != nil {
+		t.Fatalf("second ContributeFile: %v", err)
+	}
+	if len(sub.submitted) != 1 {
+		t.Fatalf("provider submissions = %d, want one", len(sub.submitted))
+	}
+	if len(first) != 1 || first[0].Status != OutcomeStatusConflict {
+		t.Fatalf("first outcomes = %+v, want conflict", first)
+	}
+	if len(second) != 1 || second[0].Status != OutcomeStatusSkipped {
+		t.Fatalf("second outcomes = %+v, want skipped", second)
+	}
+	if len(rec.recorded) != 1 || rec.recorded[0].Status != OutcomeStatusConflict || rec.recorded[0].HTTPStatus == nil || *rec.recorded[0].HTTPStatus != 409 {
+		t.Fatalf("recorded = %+v, want terminal HTTP 409 conflict", rec.recorded)
+	}
+}
+
+func TestContributeDeduplicatesSameTargetAcrossFiles(t *testing.T) {
+	sub := &fakeSubmitter{id: "introdb"}
+	firstFile := newContribFile()
+	firstFile.IntroStart, firstFile.IntroEnd = floatPtr(0), floatPtr(60)
+	firstFile.IntroMarkersSource = strPtr(models.MarkerSourceManual)
+	secondFile := *firstFile
+	secondFile.ID++
+	rec := &fakeRecorder{}
+	svc := newContribService(sub, fakeConfig{"introdb": {Provider: "introdb", ContributeEnabled: true}}, rec)
+
+	if _, err := svc.ContributeFile(context.Background(), firstFile, ContributeOptions{}); err != nil {
+		t.Fatalf("first ContributeFile: %v", err)
+	}
+	outcomes, err := svc.ContributeFile(context.Background(), &secondFile, ContributeOptions{})
+	if err != nil {
+		t.Fatalf("second ContributeFile: %v", err)
+	}
+	if len(sub.submitted) != 1 {
+		t.Fatalf("provider submissions = %d, want one for identical provider payloads", len(sub.submitted))
+	}
+	if len(outcomes) != 1 || outcomes[0].Status != OutcomeStatusSkipped {
+		t.Fatalf("second outcomes = %+v, want skipped", outcomes)
+	}
+}
+
+func TestContributeClaimsSameTargetBeforeConcurrentSubmit(t *testing.T) {
+	started := make(chan struct{})
+	release := make(chan struct{})
+	t.Cleanup(func() {
+		select {
+		case <-release:
+		default:
+			close(release)
+		}
+	})
+	sub := &fakeSubmitter{id: "introdb", started: started, release: release}
+	file := newContribFile()
+	file.IntroStart, file.IntroEnd = floatPtr(0), floatPtr(60)
+	file.IntroMarkersSource = strPtr(models.MarkerSourceManual)
+	duplicate := *file
+	duplicate.ID++
+	rec := &fakeRecorder{}
+	svc := newContribService(sub, fakeConfig{"introdb": {Provider: "introdb", ContributeEnabled: true}}, rec)
+
+	type result struct {
+		outcomes []ContributionOutcome
+		err      error
+	}
+	firstDone := make(chan result, 1)
+	go func() {
+		outcomes, err := svc.ContributeFile(context.Background(), file, ContributeOptions{})
+		firstDone <- result{outcomes: outcomes, err: err}
+	}()
+	<-started
+
+	second, err := svc.ContributeFile(context.Background(), &duplicate, ContributeOptions{})
+	if err != nil {
+		t.Fatalf("second ContributeFile: %v", err)
+	}
+	if len(second) != 1 || second[0].Status != OutcomeStatusSkipped {
+		t.Fatalf("second outcomes = %+v, want skipped while first submission is in flight", second)
+	}
+
+	close(release)
+	first := <-firstDone
+	if first.err != nil {
+		t.Fatalf("first ContributeFile: %v", first.err)
+	}
+	if len(first.outcomes) != 1 || first.outcomes[0].Status != SubmissionStatusPending {
+		t.Fatalf("first outcomes = %+v, want pending", first.outcomes)
+	}
+	if len(sub.submitted) != 1 {
+		t.Fatalf("provider submissions = %d, want one concurrent submission", len(sub.submitted))
+	}
+}
+
+func TestContributeBoundsProviderSubmissionBelowClaimLease(t *testing.T) {
+	deadlines := make(chan time.Time, 1)
+	sub := &fakeSubmitter{id: "introdb", deadline: deadlines}
+	file := newContribFile()
+	file.IntroStart, file.IntroEnd = floatPtr(0), floatPtr(60)
+	file.IntroMarkersSource = strPtr(models.MarkerSourceManual)
+	svc := newContribService(sub, fakeConfig{"introdb": {Provider: "introdb", ContributeEnabled: true}}, &fakeRecorder{})
+
+	startedAt := time.Now()
+	if _, err := svc.ContributeFile(context.Background(), file, ContributeOptions{}); err != nil {
+		t.Fatalf("ContributeFile: %v", err)
+	}
+	deadline := <-deadlines
+	if deadline.IsZero() {
+		t.Fatal("provider submission context has no deadline")
+	}
+	if got := deadline.Sub(startedAt); got < contributionSubmitLimit-time.Second || got > contributionSubmitLimit+time.Second {
+		t.Fatalf("provider deadline = %v, want approximately %v", got, contributionSubmitLimit)
+	}
+	if contributionSubmitLimit >= contributionClaimLease {
+		t.Fatalf("provider timeout %v must stay below claim lease %v", contributionSubmitLimit, contributionClaimLease)
+	}
+}
+
+func TestContributeRetriesTransientErrors(t *testing.T) {
+	sub := &fakeSubmitter{id: "introdb", err: errors.New("temporary provider failure")}
+	file := newContribFile()
+	file.IntroStart, file.IntroEnd = floatPtr(0), floatPtr(60)
+	file.IntroMarkersSource = strPtr(models.MarkerSourceManual)
+	rec := &fakeRecorder{}
+	svc := newContribService(sub, fakeConfig{"introdb": {Provider: "introdb", ContributeEnabled: true}}, rec)
+
+	for range 2 {
+		if _, err := svc.ContributeFile(context.Background(), file, ContributeOptions{}); err != nil {
+			t.Fatalf("ContributeFile: %v", err)
+		}
+	}
+	if len(sub.submitted) != 2 {
+		t.Fatalf("provider submissions = %d, want retry after transient failure", len(sub.submitted))
+	}
+}
+
+func TestContributeReleasesClaimAfterProviderCancelsRequest(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	sub := &fakeSubmitter{id: "introdb", err: context.Canceled, onSubmit: cancel}
+	file := newContribFile()
+	file.IntroStart, file.IntroEnd = floatPtr(0), floatPtr(60)
+	file.IntroMarkersSource = strPtr(models.MarkerSourceManual)
+	rec := &fakeRecorder{}
+	svc := newContribService(sub, fakeConfig{"introdb": {Provider: "introdb", ContributeEnabled: true}}, rec)
+
+	first, err := svc.ContributeFile(ctx, file, ContributeOptions{})
+	if err != nil {
+		t.Fatalf("first ContributeFile: %v", err)
+	}
+	if len(first) != 1 || first[0].Status != OutcomeStatusError {
+		t.Fatalf("first outcomes = %+v, want retryable error", first)
+	}
+
+	second, err := svc.ContributeFile(context.Background(), file, ContributeOptions{})
+	if err != nil {
+		t.Fatalf("second ContributeFile: %v", err)
+	}
+	if len(second) != 1 || second[0].Status != OutcomeStatusError {
+		t.Fatalf("second outcomes = %+v, want a retried provider error", second)
+	}
+	if len(sub.submitted) != 2 {
+		t.Fatalf("provider submissions = %d, want retry after canceled request", len(sub.submitted))
+	}
+}
+
 func TestContributeSkipsWhenProviderRequiredIDMissing(t *testing.T) {
 	sub := &fakeSubmitter{id: "introdb", required: []string{ExternalIDKeyTMDB}}
 	file := newContribFile()
@@ -262,3 +504,33 @@ func TestContentHashStableAndSensitive(t *testing.T) {
 		t.Error("changed resolved target should change hash")
 	}
 }
+
+func TestContributionClaimConflictMatchesOnlyGlobalClaimIndex(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{
+			name: "global claim conflict",
+			err:  &pgconn.PgError{Code: "23505", ConstraintName: contributionClaimIndex},
+			want: true,
+		},
+		{
+			name: "other unique conflict",
+			err:  &pgconn.PgError{Code: "23505", ConstraintName: "marker_contributions_pkey"},
+		},
+		{
+			name: "other database error",
+			err:  &pgconn.PgError{Code: "23503", ConstraintName: contributionClaimIndex},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := isContributionClaimConflict(tt.err); got != tt.want {
+				t.Fatalf("isContributionClaimConflict() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

internal/markers/contribution_repo.go

@@ -4,17 +4,23 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"strconv"
 	"time"
 
 	"github.com/Silo-Server/silo-server/internal/models"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/jackc/pgx/v5/pgxpool"
 )
 
+const contributionClaimIndex = "marker_contributions_provider_hash_active_uidx"
+
 // ContributionRow is one submission audit record from marker_contributions.
 type ContributionRow struct {
 	ID               string
+	ClaimToken       string
 	MediaFileID      int
 	Provider         string
 	SegmentKind      string
@@ -31,6 +37,13 @@ type ContributionRow struct {
 	UpdatedAt        time.Time
 }
 
+// ContributionClaim identifies one ownership generation of a contribution
+// row. The token fences a late worker from recording over a reclaimed claim.
+type ContributionClaim struct {
+	ID    string
+	Token string
+}
+
 // ContributionStore persists marker contribution attempts for idempotency and
 // audit.
 type ContributionStore struct {
@@ -61,52 +74,171 @@ func ptrIntStr(v *int64) string {
 	return strconv.FormatInt(*v, 10)
 }
 
-// AlreadySubmitted reports whether a non-error contribution with this value-hash
-// already exists for the file+provider+segment. Errors are excluded so failed
-// attempts can be retried.
-func (s *ContributionStore) AlreadySubmitted(ctx context.Context, fileID int, provider, segmentKind, contentHash string) (bool, error) {
+// Claim atomically reserves a provider-target payload before the network call.
+// A terminal or fresh in-flight row keeps the claim active; recording a
+// retryable error releases it, and a stale in-flight row can be reclaimed.
+// The advisory lock and partial unique index serialize identical payloads
+// across different local media files and server workers.
+func (s *ContributionStore) Claim(ctx context.Context, row ContributionRow, staleAfter time.Duration) (ContributionClaim, bool, error) {
 	if s == nil || s.pool == nil {
-		return false, nil
+		return ContributionClaim{}, false, fmt.Errorf("contribution store unavailable")
 	}
-	var exists bool
-	if err := s.pool.QueryRow(ctx, `
-		SELECT EXISTS (
-			SELECT 1 FROM marker_contributions
-			WHERE media_file_id = $1 AND provider = $2 AND segment_kind = $3
-			  AND content_hash = $4 AND status <> 'error'
-		)`, fileID, provider, segmentKind, contentHash).Scan(&exists); err != nil {
-		return false, fmt.Errorf("check marker contribution: %w", err)
+	if staleAfter <= 0 {
+		return ContributionClaim{}, false, fmt.Errorf("contribution claim lease must be positive")
 	}
-	return exists, nil
-}
 
-// Record upserts a contribution row keyed by (file, provider, segment, hash).
-func (s *ContributionStore) Record(ctx context.Context, row ContributionRow) error {
-	if s == nil || s.pool == nil {
-		return fmt.Errorf("contribution store unavailable")
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return ContributionClaim{}, false, fmt.Errorf("begin marker contribution claim: %w", err)
 	}
-	if _, err := s.pool.Exec(ctx, `
+	defer func() { _ = tx.Rollback(ctx) }()
+
+	if _, err := tx.Exec(ctx, `
+		SELECT pg_advisory_xact_lock(
+			hashtextextended($1 || chr(31) || $2 || chr(31) || $3, 0)
+		)`, row.Provider, row.SegmentKind, row.ContentHash); err != nil {
+		return ContributionClaim{}, false, fmt.Errorf("lock marker contribution claim: %w", err)
+	}
+
+	leaseSeconds := int64(staleAfter / time.Second)
+	if leaseSeconds < 1 {
+		leaseSeconds = 1
+	}
+	var activeID, activeStatus string
+	var stale bool
+	err = tx.QueryRow(ctx, `
+		SELECT id, status, updated_at < now() - ($4 * interval '1 second')
+		FROM marker_contributions
+		WHERE provider = $1 AND segment_kind = $2 AND content_hash = $3
+		  AND claim_active`,
+		row.Provider, row.SegmentKind, row.ContentHash, leaseSeconds,
+	).Scan(&activeID, &activeStatus, &stale)
+	if err != nil && !errors.Is(err, pgx.ErrNoRows) {
+		return ContributionClaim{}, false, fmt.Errorf("find marker contribution claim: %w", err)
+	}
+	if err == nil {
+		if activeStatus != contributionStatusClaim || !stale {
+			return ContributionClaim{}, false, nil
+		}
+		claim, reclaimed, err := reclaimContribution(ctx, tx, activeID, row, leaseSeconds)
+		if err != nil {
+			return ContributionClaim{}, false, err
+		}
+		if reclaimed {
+			if err := tx.Commit(ctx); err != nil {
+				return ContributionClaim{}, false, fmt.Errorf("commit marker contribution claim: %w", err)
+			}
+			return claim, true, nil
+		}
+	}
+
+	var claim ContributionClaim
+	err = tx.QueryRow(ctx, `
 		INSERT INTO marker_contributions (
 			media_file_id, provider, segment_kind, source,
 			submitted_start_ms, submitted_end_ms, video_duration_ms,
-			content_hash, submission_id, status, http_status, error, updated_at
-		) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12, now())
+			content_hash, status, claim_active, claim_token, updated_at
+		) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,true,gen_random_uuid(),now())
 		ON CONFLICT (media_file_id, provider, segment_kind, content_hash) DO UPDATE SET
 			source = EXCLUDED.source,
 			submitted_start_ms = EXCLUDED.submitted_start_ms,
 			submitted_end_ms = EXCLUDED.submitted_end_ms,
 			video_duration_ms = EXCLUDED.video_duration_ms,
-			submission_id = EXCLUDED.submission_id,
+			submission_id = NULL,
 			status = EXCLUDED.status,
-			http_status = EXCLUDED.http_status,
-			error = EXCLUDED.error,
-			updated_at = now()`,
+			http_status = NULL,
+			error = NULL,
+			claim_active = true,
+			claim_token = gen_random_uuid(),
+			updated_at = now()
+		WHERE NOT marker_contributions.claim_active
+		RETURNING id, claim_token`,
 		row.MediaFileID, row.Provider, row.SegmentKind, row.Source,
 		row.SubmittedStartMs, row.SubmittedEndMs, row.VideoDurationMs,
-		row.ContentHash, row.SubmissionID, row.Status, row.HTTPStatus, row.Error,
-	); err != nil {
+		row.ContentHash, contributionStatusClaim,
+	).Scan(&claim.ID, &claim.Token)
+	if errors.Is(err, pgx.ErrNoRows) || isContributionClaimConflict(err) {
+		return ContributionClaim{}, false, nil
+	}
+	if err != nil {
+		return ContributionClaim{}, false, fmt.Errorf("claim marker contribution: %w", err)
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return ContributionClaim{}, false, fmt.Errorf("commit marker contribution claim: %w", err)
+	}
+	return claim, true, nil
+}
+
+func reclaimContribution(ctx context.Context, tx pgx.Tx, id string, row ContributionRow, leaseSeconds int64) (ContributionClaim, bool, error) {
+	var claim ContributionClaim
+	err := tx.QueryRow(ctx, `
+		UPDATE marker_contributions SET
+			source = $2,
+			submitted_start_ms = $3,
+			submitted_end_ms = $4,
+			video_duration_ms = $5,
+			submission_id = NULL,
+			status = $6,
+			http_status = NULL,
+			error = NULL,
+			claim_token = gen_random_uuid(),
+			updated_at = now()
+		WHERE id = $1 AND claim_active AND status = $6
+		  AND updated_at < now() - ($7 * interval '1 second')
+		RETURNING id, claim_token`,
+		id, row.Source, row.SubmittedStartMs, row.SubmittedEndMs,
+		row.VideoDurationMs, contributionStatusClaim, leaseSeconds,
+	).Scan(&claim.ID, &claim.Token)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return ContributionClaim{}, false, nil
+	}
+	if err != nil {
+		return ContributionClaim{}, false, fmt.Errorf("reclaim marker contribution: %w", err)
+	}
+	return claim, true, nil
+}
+
+func isContributionClaimConflict(err error) bool {
+	var pgErr *pgconn.PgError
+	return errors.As(err, &pgErr) &&
+		pgErr.Code == "23505" &&
+		pgErr.ConstraintName == contributionClaimIndex
+}
+
+// Record completes a claimed contribution. Retryable errors release the
+// provider-target claim; every other result preserves it for deduplication.
+func (s *ContributionStore) Record(ctx context.Context, row ContributionRow) error {
+	if s == nil || s.pool == nil {
+		return fmt.Errorf("contribution store unavailable")
+	}
+	if row.ID == "" || row.ClaimToken == "" {
+		return fmt.Errorf("record marker contribution: claim identity missing")
+	}
+	claimActive := row.Status != OutcomeStatusError
+	tag, err := s.pool.Exec(ctx, `
+		UPDATE marker_contributions SET
+			source = $3,
+			submitted_start_ms = $4,
+			submitted_end_ms = $5,
+			video_duration_ms = $6,
+			submission_id = $7,
+			status = $8,
+			http_status = $9,
+			error = $10,
+			claim_active = $11,
+			updated_at = now()
+		WHERE id = $1 AND claim_token = $2 AND claim_active`,
+		row.ID, row.ClaimToken, row.Source,
+		row.SubmittedStartMs, row.SubmittedEndMs, row.VideoDurationMs,
+		row.SubmissionID, row.Status, row.HTTPStatus, row.Error,
+		claimActive,
+	)
+	if err != nil {
 		return fmt.Errorf("record marker contribution: %w", err)
 	}
+	if tag.RowsAffected() != 1 {
+		return fmt.Errorf("record marker contribution: claim not found")
+	}
 	return nil
 }
 

internal/markers/contribution_repo_test.go

@@ -0,0 +1,202 @@
+package markers
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+type contributionStoreFixture struct {
+	store    *ContributionStore
+	pool     *pgxpool.Pool
+	provider string
+	fileIDs  [2]int
+}
+
+func newContributionStoreFixture(t *testing.T) contributionStoreFixture {
+	t.Helper()
+	dsn := os.Getenv("SILO_TEST_DATABASE_URL")
+	if dsn == "" {
+		t.Skip("SILO_TEST_DATABASE_URL is not set")
+	}
+	ctx := context.Background()
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Fatalf("connect test database: %v", err)
+	}
+	t.Cleanup(pool.Close)
+
+	var migrated bool
+	if err := pool.QueryRow(ctx, `
+		SELECT to_regclass('public.marker_contributions_provider_hash_active_uidx') IS NOT NULL
+		   AND EXISTS (
+			SELECT 1 FROM information_schema.columns
+			WHERE table_schema = 'public' AND table_name = 'marker_contributions'
+			  AND column_name = 'claim_token'
+		)`).Scan(&migrated); err != nil {
+		t.Fatalf("check contribution claim migration: %v", err)
+	}
+	if !migrated {
+		t.Skip("marker contribution claim migration has not been applied")
+	}
+
+	suffix := time.Now().UnixNano()
+	provider := fmt.Sprintf("claim-test-%d", suffix)
+	var folderID int
+	if err := pool.QueryRow(ctx, `
+		INSERT INTO media_folders (type, name)
+		VALUES ('shows', $1)
+		RETURNING id`, provider).Scan(&folderID); err != nil {
+		t.Fatalf("seed media folder: %v", err)
+	}
+	var fileIDs [2]int
+	for i := range fileIDs {
+		if err := pool.QueryRow(ctx, `
+			INSERT INTO media_files (media_folder_id, file_path)
+			VALUES ($1, $2)
+			RETURNING id`, folderID, fmt.Sprintf("/claim-test/%d-%d.mkv", suffix, i)).Scan(&fileIDs[i]); err != nil {
+			t.Fatalf("seed media file: %v", err)
+		}
+	}
+	t.Cleanup(func() {
+		_, _ = pool.Exec(ctx, `DELETE FROM marker_contributions WHERE provider = $1`, provider)
+		_, _ = pool.Exec(ctx, `DELETE FROM media_files WHERE id = ANY($1)`, fileIDs[:])
+		_, _ = pool.Exec(ctx, `DELETE FROM media_folders WHERE id = $1`, folderID)
+	})
+
+	return contributionStoreFixture{
+		store:    NewContributionStore(pool),
+		pool:     pool,
+		provider: provider,
+		fileIDs:  fileIDs,
+	}
+}
+
+func (f contributionStoreFixture) row(fileID int, hash string) ContributionRow {
+	start, end, duration := int64(0), int64(60_000), int64(1_800_000)
+	return ContributionRow{
+		MediaFileID:      fileID,
+		Provider:         f.provider,
+		SegmentKind:      "intro",
+		Source:           "manual",
+		SubmittedStartMs: &start,
+		SubmittedEndMs:   &end,
+		VideoDurationMs:  &duration,
+		ContentHash:      hash,
+		Status:           contributionStatusClaim,
+	}
+}
+
+func expireContributionClaim(t *testing.T, fixture contributionStoreFixture, id string) {
+	t.Helper()
+	if _, err := fixture.pool.Exec(context.Background(), `
+		UPDATE marker_contributions
+		SET updated_at = now() - interval '1 hour'
+		WHERE id = $1`, id); err != nil {
+		t.Fatalf("expire contribution claim: %v", err)
+	}
+}
+
+func TestContributionStoreRecoversStaleClaimAcrossFiles(t *testing.T) {
+	fixture := newContributionStoreFixture(t)
+	ctx := context.Background()
+	firstRow := fixture.row(fixture.fileIDs[0], "cross-file-stale")
+	first, claimed, err := fixture.store.Claim(ctx, firstRow, contributionClaimLease)
+	if err != nil || !claimed {
+		t.Fatalf("first Claim = (%+v, %v, %v), want claimed", first, claimed, err)
+	}
+
+	secondRow := fixture.row(fixture.fileIDs[1], firstRow.ContentHash)
+	if _, claimed, err := fixture.store.Claim(ctx, secondRow, contributionClaimLease); err != nil || claimed {
+		t.Fatalf("fresh duplicate Claim = (claimed=%v, %v), want not claimed", claimed, err)
+	}
+
+	expireContributionClaim(t, fixture, first.ID)
+	second, claimed, err := fixture.store.Claim(ctx, secondRow, contributionClaimLease)
+	if err != nil || !claimed {
+		t.Fatalf("stale cross-file Claim = (%+v, %v, %v), want claimed", second, claimed, err)
+	}
+	if second.ID != first.ID || second.Token == first.Token {
+		t.Fatalf("reclaimed claim = %+v, want row %s with a fresh token", second, first.ID)
+	}
+
+	staleResult := firstRow
+	staleResult.ID, staleResult.ClaimToken = first.ID, first.Token
+	staleResult.Status = OutcomeStatusError
+	if err := fixture.store.Record(ctx, staleResult); err == nil {
+		t.Fatal("stale worker recorded over reclaimed claim")
+	}
+
+	terminal := secondRow
+	terminal.ID, terminal.ClaimToken = second.ID, second.Token
+	terminal.Status = OutcomeStatusConflict
+	if err := fixture.store.Record(ctx, terminal); err != nil {
+		t.Fatalf("record current claim: %v", err)
+	}
+	expireContributionClaim(t, fixture, second.ID)
+	if _, claimed, err := fixture.store.Claim(ctx, firstRow, contributionClaimLease); err != nil || claimed {
+		t.Fatalf("terminal Claim = (claimed=%v, %v), want permanently blocked", claimed, err)
+	}
+}
+
+func TestContributionStoreRecoversStaleClaimForSameFile(t *testing.T) {
+	fixture := newContributionStoreFixture(t)
+	ctx := context.Background()
+	row := fixture.row(fixture.fileIDs[0], "same-file-stale")
+	first, claimed, err := fixture.store.Claim(ctx, row, contributionClaimLease)
+	if err != nil || !claimed {
+		t.Fatalf("first Claim = (%+v, %v, %v), want claimed", first, claimed, err)
+	}
+	expireContributionClaim(t, fixture, first.ID)
+	second, claimed, err := fixture.store.Claim(ctx, row, contributionClaimLease)
+	if err != nil || !claimed || second.ID != first.ID || second.Token == first.Token {
+		t.Fatalf("same-file stale Claim = (%+v, %v, %v), want reclaimed row with fresh token", second, claimed, err)
+	}
+}
+
+func TestContributionStoreAllowsOnlyOneConcurrentStaleTakeover(t *testing.T) {
+	fixture := newContributionStoreFixture(t)
+	ctx := context.Background()
+	row := fixture.row(fixture.fileIDs[0], "concurrent-stale")
+	first, claimed, err := fixture.store.Claim(ctx, row, contributionClaimLease)
+	if err != nil || !claimed {
+		t.Fatalf("first Claim = (%+v, %v, %v), want claimed", first, claimed, err)
+	}
+	expireContributionClaim(t, fixture, first.ID)
+
+	const workers = 12
+	results := make(chan bool, workers)
+	errs := make(chan error, workers)
+	var ready sync.WaitGroup
+	ready.Add(workers)
+	start := make(chan struct{})
+	for range workers {
+		go func() {
+			ready.Done()
+			<-start
+			_, claimed, err := fixture.store.Claim(ctx, row, contributionClaimLease)
+			results <- claimed
+			errs <- err
+		}()
+	}
+	ready.Wait()
+	close(start)
+
+	claimedCount := 0
+	for range workers {
+		if err := <-errs; err != nil {
+			t.Fatalf("concurrent Claim: %v", err)
+		}
+		if <-results {
+			claimedCount++
+		}
+	}
+	if claimedCount != 1 {
+		t.Fatalf("concurrent stale claims won = %d, want 1", claimedCount)
+	}
+}

internal/markers/plugin_provider.go

@@ -3,6 +3,7 @@ package markers
 import (
 	"context"
 	"fmt"
+	"net/http"
 	"strconv"
 	"strings"
 	"time"
@@ -367,6 +368,18 @@ func pluginProviderError(providerID string, err error) error {
 		return nil
 	}
 	st, ok := status.FromError(err)
+	message := err.Error()
+	if ok {
+		message = st.Message()
+	}
+	upstreamStatus, legacyHTTPConflict := pluginSubmissionHTTPStatus(message)
+	if (ok && st.Code() == codes.AlreadyExists) || (legacyHTTPConflict && upstreamStatus == http.StatusConflict) {
+		return &SubmissionConflictError{
+			Provider:   providerID,
+			HTTPStatus: http.StatusConflict,
+			Message:    err.Error(),
+		}
+	}
 	if !ok || st.Code() != codes.ResourceExhausted {
 		return err
 	}
@@ -380,6 +393,26 @@ func pluginProviderError(providerID string, err error) error {
 	return &RetryAfterError{Provider: providerID, RetryAfter: retryAfter, Message: err.Error()}
 }
 
+// pluginSubmissionHTTPStatus recognizes the legacy marker-provider error text
+// used before plugins could expose an AlreadyExists gRPC status. Keep this
+// narrow to submit errors so unrelated provider failures are not reclassified.
+func pluginSubmissionHTTPStatus(message string) (int, bool) {
+	const marker = "submit HTTP "
+	start := strings.Index(message, marker)
+	if start < 0 {
+		return 0, false
+	}
+	tail := strings.TrimSpace(message[start+len(marker):])
+	if end := strings.IndexAny(tail, ": "); end >= 0 {
+		tail = tail[:end]
+	}
+	code, err := strconv.Atoi(tail)
+	if err != nil || code < 100 || code > 599 {
+		return 0, false
+	}
+	return code, true
+}
+
 func PluginRequiredExternalIDsFromMetadata(metadata map[string]any) []string {
 	raw, ok := metadata["required_external_ids"]
 	if !ok {

internal/markers/plugin_provider_test.go

@@ -2,17 +2,21 @@ package markers
 
 import (
 	"context"
+	"errors"
 	"testing"
 	"time"
 
 	pluginv1 "github.com/Silo-Server/silo-plugin-sdk/pkg/pluginproto/silo/plugin/v1"
 	"github.com/Silo-Server/silo-server/internal/models"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 type fakePluginMarkerClient struct {
 	fetchResp *pluginv1.FetchMarkersResponse
 	fetchReq  *pluginv1.FetchMarkersRequest
 	submitReq *pluginv1.SubmitMarkerRequest
+	submitErr error
 }
 
 func (f *fakePluginMarkerClient) FetchMarkers(_ context.Context, req *pluginv1.FetchMarkersRequest) (*pluginv1.FetchMarkersResponse, error) {
@@ -22,6 +26,9 @@ func (f *fakePluginMarkerClient) FetchMarkers(_ context.Context, req *pluginv1.F
 
 func (f *fakePluginMarkerClient) SubmitMarker(_ context.Context, req *pluginv1.SubmitMarkerRequest) (*pluginv1.SubmitMarkerResponse, error) {
 	f.submitReq = req
+	if f.submitErr != nil {
+		return nil, f.submitErr
+	}
 	return &pluginv1.SubmitMarkerResponse{SubmissionId: "sub1", Status: SubmissionStatusPending, Weight: 2}, nil
 }
 
@@ -159,3 +166,71 @@ func TestPluginProviderSubmitMapsRequest(t *testing.T) {
 		t.Fatalf("submit request segment = %+v", client.submitReq)
 	}
 }
+
+func TestPluginProviderSubmitMapsConflicts(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+	}{
+		{
+			name: "structured already exists",
+			err:  status.Error(codes.AlreadyExists, "submission already exists"),
+		},
+		{
+			name: "legacy HTTP 409",
+			err:  status.Error(codes.Unknown, `introdb: submit HTTP 409: {"error":"already submitted"}`),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			client := &fakePluginMarkerClient{submitErr: tt.err}
+			provider, err := NewPluginProviderWithClientFactory(PluginProviderOptions{
+				InstallationID: 12,
+				CapabilityID:   "markers",
+			}, func(context.Context, int, string) (pluginMarkerClient, error) {
+				return client, nil
+			})
+			if err != nil {
+				t.Fatalf("NewPluginProviderWithClientFactory: %v", err)
+			}
+
+			_, err = provider.SubmitMarker(context.Background(), SubmissionRequest{
+				Kind:        ItemKindMovie,
+				ExternalIDs: map[string]string{ExternalIDKeyTMDB: "123"},
+				Segment:     MarkerKindIntro,
+			})
+			var conflict *SubmissionConflictError
+			if !errors.As(err, &conflict) {
+				t.Fatalf("error = %T %v, want SubmissionConflictError", err, err)
+			}
+			if conflict.Provider != "plugin:12:markers" || conflict.HTTPStatus != 409 {
+				t.Fatalf("conflict = %+v", conflict)
+			}
+		})
+	}
+}
+
+func TestPluginProviderSubmitKeepsOtherErrorsRetryable(t *testing.T) {
+	wantErr := status.Error(codes.Unknown, "introdb: submit HTTP 500: unavailable")
+	client := &fakePluginMarkerClient{submitErr: wantErr}
+	provider, err := NewPluginProviderWithClientFactory(PluginProviderOptions{
+		InstallationID: 12,
+		CapabilityID:   "markers",
+	}, func(context.Context, int, string) (pluginMarkerClient, error) {
+		return client, nil
+	})
+	if err != nil {
+		t.Fatalf("NewPluginProviderWithClientFactory: %v", err)
+	}
+
+	_, err = provider.SubmitMarker(context.Background(), SubmissionRequest{
+		Kind:        ItemKindMovie,
+		ExternalIDs: map[string]string{ExternalIDKeyTMDB: "123"},
+		Segment:     MarkerKindIntro,
+	})
+	var conflict *SubmissionConflictError
+	if errors.As(err, &conflict) {
+		t.Fatalf("error = %+v, want retryable provider error", conflict)
+	}
+}

internal/markers/types.go

@@ -145,6 +145,28 @@ type SubmissionResult struct {
 	Weight float64
 }
 
+// SubmissionConflictError marks a provider refusal that cannot succeed when
+// the exact same payload is retried. A changed marker produces a different
+// content hash and remains eligible for a later submission.
+type SubmissionConflictError struct {
+	Provider   string
+	HTTPStatus int
+	Message    string
+}
+
+func (e *SubmissionConflictError) Error() string {
+	if e == nil {
+		return ""
+	}
+	if e.Message != "" {
+		return e.Message
+	}
+	if e.Provider != "" {
+		return fmt.Sprintf("%s: submission conflict", e.Provider)
+	}
+	return "submission conflict"
+}
+
 // UserStats is a contribution-account summary used to validate a key and show
 // contribution totals in the admin UI.
 type UserStats struct {

internal/taskmanager/tasks/contribute_markers.go

@@ -119,7 +119,7 @@ func (t *ContributeMarkersTask) Execute(ctx context.Context, progress taskmanage
 			}
 			for _, o := range outcomes {
 				switch o.Status {
-				case markers.OutcomeStatusSkipped:
+				case markers.OutcomeStatusSkipped, markers.OutcomeStatusConflict:
 					skipped++
 				case markers.OutcomeStatusRateLimited:
 					failed++

internal/taskmanager/tasks/contribute_markers_test.go

@@ -116,3 +116,22 @@ func TestContributeMarkersTaskStopsOnRateLimit(t *testing.T) {
 		t.Fatalf("retry_after_seconds = %d, want 90", data["retry_after_seconds"])
 	}
 }
+
+func TestContributeMarkersTaskCountsConflictAsSkipped(t *testing.T) {
+	runner := &fakeContribRunner{outcomes: []markers.ContributionOutcome{{Status: markers.OutcomeStatusConflict}}}
+	cands := &fakeCandidates{ids: []int{10}}
+	cfg := fakeAutoConfig{{Provider: "introdb", ContributeEnabled: true, ContributeAutoLocal: true}}
+	task := NewContributeMarkersTask(runner, cfg, cands, fakeFileLoader{})
+
+	prog := &contribTestProgress{}
+	if err := task.Execute(context.Background(), prog); err != nil {
+		t.Fatalf("Execute: %v", err)
+	}
+	var data map[string]int
+	if err := json.Unmarshal(prog.data, &data); err != nil {
+		t.Fatalf("decode result data: %v", err)
+	}
+	if data["submitted"] != 0 || data["skipped"] != 1 || data["failed"] != 0 {
+		t.Fatalf("result = %v, want one skipped conflict", data)
+	}
+}

migrations/sql/20260807140637_settle_marker_contribution_conflicts.sql

@@ -0,0 +1,89 @@
+-- +goose NO TRANSACTION
+
+-- +goose Up
+ALTER TABLE public.marker_contributions
+    ADD COLUMN IF NOT EXISTS claim_active boolean NOT NULL DEFAULT false,
+    ADD COLUMN IF NOT EXISTS claim_token uuid;
+
+-- Legacy marker-provider plugins reported upstream HTTP conflicts as generic
+-- errors. These conflicts are terminal for an unchanged contribution payload,
+-- so settle existing rows before the next daily task can retry them.
+UPDATE public.marker_contributions
+SET status = 'conflict',
+    http_status = 409,
+    updated_at = now()
+WHERE status = 'error'
+  AND (
+      http_status = 409
+      OR error LIKE '%submit HTTP 409:%'
+  );
+
+-- Retain every audit row while choosing one active claim for each historical
+-- provider payload. Error rows stay inactive so a later attempt can retry.
+UPDATE public.marker_contributions
+SET claim_active = false
+WHERE status = 'error';
+
+WITH ranked AS (
+    SELECT id,
+           ROW_NUMBER() OVER (
+               PARTITION BY provider, segment_kind, content_hash
+               ORDER BY claim_active DESC, updated_at DESC, submitted_at DESC, id
+           ) AS row_number
+    FROM public.marker_contributions
+    WHERE status <> 'error'
+)
+UPDATE public.marker_contributions AS contribution
+SET claim_active = false
+FROM ranked
+WHERE contribution.id = ranked.id
+  AND ranked.row_number > 1;
+
+WITH ranked AS (
+    SELECT id,
+           ROW_NUMBER() OVER (
+               PARTITION BY provider, segment_kind, content_hash
+               ORDER BY claim_active DESC, updated_at DESC, submitted_at DESC, id
+           ) AS row_number
+    FROM public.marker_contributions
+    WHERE status <> 'error'
+)
+UPDATE public.marker_contributions AS contribution
+SET claim_active = true
+FROM ranked
+WHERE contribution.id = ranked.id
+  AND ranked.row_number = 1
+  AND NOT contribution.claim_active;
+
+-- Remove an INVALID remnant before retrying an interrupted concurrent build.
+-- +goose StatementBegin
+DO $$
+BEGIN
+    IF EXISTS (
+        SELECT 1
+        FROM pg_class c
+        JOIN pg_namespace n ON n.oid = c.relnamespace
+        JOIN pg_index i ON i.indexrelid = c.oid
+        WHERE n.nspname = 'public'
+          AND c.relname = 'marker_contributions_provider_hash_active_uidx'
+          AND NOT i.indisvalid
+    ) THEN
+        DROP INDEX public.marker_contributions_provider_hash_active_uidx;
+    END IF;
+END;
+$$;
+-- +goose StatementEnd
+
+-- The active claim is provider-payload scoped rather than local-file scoped.
+-- Build the global uniqueness guarantee without blocking contribution writes.
+CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS marker_contributions_provider_hash_active_uidx
+    ON public.marker_contributions (provider, segment_kind, content_hash)
+    WHERE claim_active;
+
+-- +goose Down
+DROP INDEX CONCURRENTLY IF EXISTS public.marker_contributions_provider_hash_active_uidx;
+ALTER TABLE public.marker_contributions
+    DROP COLUMN IF EXISTS claim_token,
+    DROP COLUMN IF EXISTS claim_active;
+-- Settled conflict rows are intentionally retained: reverting them to generic
+-- errors would make the server resubmit them on the next scheduled run.