verification

Documentation

Overview

Package verification provides the durable, lease-based post-upload verification service for the process-wide upload architecture (issue 184).

It is the primary post-check path: instead of holding a queue upload slot during propagation delay, completed transfers are recorded durably and this service verifies them in the background. It streams a file's manifest, issues STAT requests, and persists ONLY the articles that fail. Failed articles are then re-posted (reusing their exact Message-ID and headers from the manifest) or STAT-rechecked with backoff, using database leases so the work survives process restarts.

The service is decoupled from the upload/NNTP layers through the Stater and Reposter interfaces so it can be unit-tested in isolation and wired into the processor later.

Index

• type Cleaner
• type Config
• type Reposter
• type Service
  • func New(store *transferstore.Store, stater Stater, reposter Reposter, cfg Config, ...) *Service
  • func (s *Service) ProcessDueFailures(ctx context.Context, now time.Time) (int, error)
  • func (s *Service) Run(ctx context.Context)
  • func (s *Service) SetCleaner(c Cleaner)
  • func (s *Service) VerifyDueFiles(ctx context.Context, now time.Time, limit int) (int, error)
  • func (s *Service) VerifyFile(ctx context.Context, tf transferstore.TransferFile) error
• type Stater

Types

type Cleaner

type Cleaner interface {
	CleanupTransfer(ctx context.Context, transferID string) (bool, error)
}

Cleaner runs post-verification cleanup for a transfer once it is fully verified (deletes originals per policy, removes PAR2 and manifests). It is a no-op while the transfer is not yet terminal, and retains everything on failure.

type Config

type Config struct {
	// MaxConcurrentChecks bounds simultaneous STAT requests. 0 = auto (16).
	MaxConcurrentChecks int
	// StatBatchSize is the number of message-IDs checked per StatBatch call
	// during file verification. 0 = 100.
	StatBatchSize int
	// PropagationDelay is how long to wait after a (re)post before checking.
	PropagationDelay time.Duration
	// MaxReposts is the maximum number of times a missing article is re-posted
	// before falling back to STAT-only deferred checking.
	MaxReposts int
	// DeferredBackoff is the base delay between STAT-only rechecks; it grows
	// exponentially up to MaxBackoff.
	DeferredBackoff time.Duration
	MaxBackoff      time.Duration
	// MaxDeferredChecks caps STAT-only rechecks before the article (and its
	// file) is marked permanently failed.
	MaxDeferredChecks int
	// LeaseDuration is how long a claimed batch is owned before it can be
	// reclaimed by another worker after a crash.
	LeaseDuration time.Duration
	// BatchSize is the maximum number of failures claimed per cycle.
	BatchSize int
	// PollInterval is how often the Run loop checks for due files/failures.
	PollInterval time.Duration
}

Config controls verification timing and concurrency. Zero values fall back to conservative defaults via withDefaults.

type Reposter

type Reposter interface {
	Repost(ctx context.Context, rec manifest.ArticleRecord) error
}

Reposter re-posts a single article described by its manifest record, reusing the record's Message-ID and headers so the NZB stays correct.

type Service

type Service struct {
	// contains filtered or unexported fields
}

Service runs durable verification against a transfer store.

func New

func New(store *transferstore.Store, stater Stater, reposter Reposter, cfg Config, owner string) *Service

New creates a verification service. owner identifies this worker for lease ownership (e.g. a hostname+pid string).

func (*Service) ProcessDueFailures

func (s *Service) ProcessDueFailures(ctx context.Context, now time.Time) (int, error)

ProcessDueFailures claims a batch of due verification failures and handles each: re-post (while under MaxReposts) or STAT-only recheck with exponential backoff. Returns the number of failures processed. Resolving the last failure for a file flips that file to verified; exhausting retries flips it to verification_failed.

func (*Service) Run

func (s *Service) Run(ctx context.Context)

Run drives verification until ctx is cancelled: on each tick it reclaims expired leases, runs the first verification check for any due files, then processes due verification failures (re-posts/rechecks). It is intended to be started once as a background goroutine, owned by the processor.

func (*Service) SetCleaner

func (s *Service) SetCleaner(c Cleaner)

SetCleaner installs the post-verification cleaner invoked when a transfer becomes fully verified. Optional; nil disables cleanup.

func (*Service) VerifyDueFiles

func (s *Service) VerifyDueFiles(ctx context.Context, now time.Time, limit int) (int, error)

VerifyDueFiles runs the first verification check for every file whose check is due (verification_state=uploaded, next_check_at<=now), up to limit. A failure verifying one file is logged and does not abort the batch. Returns the number of files verified.

func (*Service) VerifyFile

func (s *Service) VerifyFile(ctx context.Context, tf transferstore.TransferFile) error

VerifyFile streams a transfer file's manifest, STATs every article with bounded concurrency, and persists only the misses. If no article is missing the file is marked verified; otherwise it is marked verifying and the misses are left for ProcessDueFailures to re-post/recheck.

type Stater

type Stater interface {
	Stat(ctx context.Context, messageID string) error
	StatBatch(ctx context.Context, messageIDs []string) (missing map[string]struct{}, err error)
}

Stater reports whether articles exist on the server. For Stat, a nil error means the article was found; any error means it is (still) missing. For StatBatch, the returned set contains the message-IDs NOT confirmed present (any per-ID error counts as missing).