s3-orchestrator

worker

import "github.com/afreidah/s3-orchestrator/internal/worker"

Index

Constants

Worker name labels used by admission-rejection metric and other worker-scoped telemetry. Constants prevent the silent label drift that would otherwise split a metric’s time series in Grafana when a typo lands.

const (
    WorkerNameCleanup         = "cleanup"
    WorkerNameOverReplication = "over_replication"
    WorkerNamePendingReaper   = "pending_reaper"
    WorkerNameRebalancer      = "rebalancer"
    WorkerNameReplicator      = "replicator"
)

DefaultCleanupQueueTick is the cleanup-queue worker’s per-tick cadence when the config does not override it.

const DefaultCleanupQueueTick = 1 * time.Minute

DefaultOverReplicationTick is the over-replication cleaner’s per-tick cadence when the config does not specify one.

const DefaultOverReplicationTick = 5 * time.Minute

DefaultPendingReaperTick is the pending reaper’s per-tick cadence when the config does not specify one.

const DefaultPendingReaperTick = 1 * time.Minute

DefaultRebalanceInterval is the rebalancer’s per-tick cadence when the config does not specify one.

const DefaultRebalanceInterval = 6 * time.Hour

DefaultReplicatorTick is the replicator’s per-tick cadence when the config does not specify one.

const DefaultReplicatorTick = 5 * time.Minute

DefaultScrubberInterval is the integrity scrubber’s per-tick cadence when the integrity config does not specify one.

const DefaultScrubberInterval = 6 * time.Hour

func CleanupBackoff

func CleanupBackoff(attempts int32) time.Duration

CleanupBackoff returns the backoff duration for the given attempt number. Uses exponential backoff: min(1m * 2^attempts, 24h). Short-circuits the shift for attempts >= 11 (where the doubling already exceeds the cap) and for negative inputs, since shifting by a negative or out-of-range count is undefined in Go.

func ExceedsThreshold

func ExceedsThreshold(stats map[string]core.QuotaStat, order []string, threshold float64) bool

ExceedsThreshold reports whether the utilization spread across backends (max ratio minus min ratio) is at least the configured threshold.

func NewCleanupQueueService

func NewCleanupQueueService(cleanup *CleanupWorker, locker tickrunner.AdvisoryLocker) lifecycle.Runner

NewCleanupQueueService constructs the cleanup-queue background service.

func NewOverReplicationService

func NewOverReplicationService(manager tickrunner.QuotaMetricsRefresher, overRep *OverReplicationCleaner, locker tickrunner.AdvisoryLocker) lifecycle.Runner

NewOverReplicationService constructs the over-replication cleanup service.

func NewPendingReaperService

func NewPendingReaperService(reaper *PendingReaper, locker tickrunner.AdvisoryLocker, tick time.Duration) lifecycle.Runner

NewPendingReaperService constructs the pending-objects reaper background service. The reaper resolves abandoned PUT intents by HEADing the destination backend and either promoting the intent into object_locations (bytes present) or dropping it (bytes absent). Returns nil when no pending reaper is configured.

func NewRebalancerService

func NewRebalancerService(manager tickrunner.QuotaMetricsRefresher, rebalancer *Rebalancer, locker tickrunner.AdvisoryLocker) lifecycle.Runner

NewRebalancerService constructs the rebalancer background service.

func NewReconcileService

func NewReconcileService(reconciler *Reconciler, locker tickrunner.AdvisoryLocker, interval time.Duration) lifecycle.Runner

NewReconcileService constructs the reconcile background service.

func NewReplicatorService

func NewReplicatorService(manager tickrunner.QuotaMetricsRefresher, replicator *Replicator, locker tickrunner.AdvisoryLocker) lifecycle.Runner

NewReplicatorService constructs the replication background service.

func NewScrubberService

func NewScrubberService(scrubber *Scrubber, locker tickrunner.AdvisoryLocker) lifecycle.Runner

NewScrubberService constructs the integrity scrubber background service.

func WithAdmission

func WithAdmission(ctx context.Context, ac AdmissionControl, name string, fn func())

WithAdmission acquires an admission slot, runs fn if granted, and releases the slot afterward. When admission is rejected, increments the per-worker rejection counter and returns without invoking fn. name must be one of the WorkerName* constants above.

type AdmissionControl

AdmissionControl gates concurrent access to backends.

type AdmissionControl interface {
    AcquireAdmission(ctx context.Context) bool
    ReleaseAdmission()
}

type BackendAccess

BackendAccess provides backend fleet discovery and drain-awareness.

type BackendAccess interface {
    Backends() map[string]backend.ObjectBackend
    BackendOrder() []string
    IsDraining(name string) bool
    ExcludeDraining(eligible []string) []string
}

type BackendSyncer

BackendSyncer is the interface the reconciler needs from the proxy layer. Defined here to avoid a worker->proxy import cycle.

type BackendSyncer interface {
    SyncBackend(ctx context.Context, backendName, bucket string, knownBuckets []string) (imported, skipped int, err error)
    ReconcileBackend(ctx context.Context, backendName, bucket string, knownBuckets []string) (*ReconcileResult, error)
    UpdateQuotaMetrics(ctx context.Context) error
    ReconcileUsage(ctx context.Context) (map[string]int64, error)
    BackendOrder() []string
}

type BatchRunner

BatchRunner drives one worker cycle. Concurrency is passed through to workerpool.Run (1 = sequential); Key, when set, supplies each item’s progress label; Observer, when set, receives the per-item progress steps.

type BatchRunner[T any] struct {
    Name        string
    Log         *slog.Logger
    Concurrency int
    Observer    progress.Observer
    Key         func(T) string
}

func (BatchRunner[T]) Run

func (r BatchRunner[T]) Run(ctx context.Context, items []T, fn func(context.Context, T) ItemResult) WorkSummary

Run processes items through fn with the configured concurrency, brackets each in a progress step, and returns the tallied WorkSummary. Items not reached because ctx was cancelled mid-cycle are left out of the tally, so Planned - (Attempted + Skipped) reflects the cancelled remainder.

type CleanupOps

CleanupOps is the dependency contract for CleanupWorker. It omits BackendAccess because cleanup operates on specific named backends.

type CleanupOps interface {
    AdmissionControl
    DataMover
    UsageAccessor
    RecorderProvider
}

type CleanupWorker

CleanupWorker processes the retry queue for failed object deletions.

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

func NewCleanupWorker

func NewCleanupWorker(deps CleanupWorkerDeps) *CleanupWorker

NewCleanupWorker creates a CleanupWorker with the given dependencies.

func (*CleanupWorker) ProcessCleanupQueue

func (w *CleanupWorker) ProcessCleanupQueue(ctx context.Context) WorkSummary

ProcessCleanupQueue fetches pending cleanup items and attempts to delete the orphaned objects from their respective backends.

type CleanupWorkerDeps

CleanupWorkerDeps groups the cleanup worker’s constructor parameters. InstanceID is stamped into cleanup_queue.claimed_by for observability; ClaimGracePeriod is the threshold past which an outstanding claim becomes reclaimable by another worker tick (typically 5m).

type CleanupWorkerDeps struct {
    Ops              CleanupOps
    Store            CleanupWorkerStore
    Concurrency      int
    InstanceID       string
    ClaimGracePeriod time.Duration
}

type CleanupWorkerStore

CleanupWorkerStore is the narrow persistence surface the cleanup worker needs: the cleanup queue + DLQ operations. Declared locally so the worker does not pull in the full MetadataStore.

type CleanupWorkerStore interface {
    core.CleanupStore
}

type DataMover

DataMover provides object data movement primitives.

type DataMover interface {
    GetBackend(name string) (backend.ObjectBackend, error)
    WithTimeout(ctx context.Context) (context.Context, context.CancelFunc)
    GetWithTimeout(ctx context.Context, be backend.ObjectBackend, key, rangeHeader string) (*backend.GetObjectResult, context.CancelFunc, error)
    HeadWithTimeout(ctx context.Context, be backend.ObjectBackend, key string) (*backend.HeadObjectResult, error)
    StreamCopy(ctx context.Context, src, dst backend.ObjectBackend, key string) error
    DeleteWithTimeout(ctx context.Context, be backend.ObjectBackend, key string) error
}

type ItemOutcome

ItemOutcome classifies how a single work item finished, for the batch tally.

type ItemOutcome int

const (
    // ItemSkipped marks an item the worker declined before doing any work. It
    // is the zero value, so an item whose per-item function never ran (e.g.
    // admission blocked it) counts as skipped rather than succeeded.
    ItemSkipped ItemOutcome = iota
    // ItemSucceeded marks an item the worker processed successfully.
    ItemSucceeded
    // ItemFailed marks an item the worker attempted but could not complete.
    ItemFailed
)

type ItemResult

ItemResult is what a per-item function returns: the outcome for the tally and a human-readable status for the progress stream (ignored when no observer is attached).

type ItemResult struct {
    Outcome ItemOutcome
    Status  string
}

type Ops

Ops combines fleet access, admission control, data movement, and usage tracking. Used by Rebalancer, Replicator, and OverReplicationCleaner.

type Ops interface {
    BackendAccess
    AdmissionControl
    DataMover
    UsageAccessor
    RecorderProvider
}

type OverReplicationCleaner

OverReplicationCleaner removes excess copies of objects that exceed the configured replication factor. Embeds *backendCore for access to shared infrastructure.

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

func NewOverReplicationCleaner

func NewOverReplicationCleaner(ops Ops, placement Placement, store OverReplicationCleanerStore) *OverReplicationCleaner

NewOverReplicationCleaner creates a cleaner with fleet operations, write-path placement, and a metadata store.

func (*OverReplicationCleaner) Clean

func (c *OverReplicationCleaner) Clean(ctx context.Context, cfg config.ReplicationConfig, observer progress.Observer) (int, error)

Clean finds over-replicated objects and removes excess copies to reach the target replication factor. Returns the number of copies removed. observer, when non-nil, receives a start and end step per object cleaned.

func (*OverReplicationCleaner) Config

func (c *OverReplicationCleaner) Config() *config.ReplicationConfig

Config returns the current replication configuration.

func (*OverReplicationCleaner) CountPending

func (c *OverReplicationCleaner) CountPending(ctx context.Context, factor int) (int64, error)

CountPending returns the number of objects exceeding the replication factor.

func (*OverReplicationCleaner) ScoreCopy

func (c *OverReplicationCleaner) ScoreCopy(loc *core.ObjectLocation, stats map[string]core.QuotaStat) float64

ScoreCopy assigns a retention score to a copy based on its backend’s state:

  • draining backend: 0 (always remove first)
  • circuit-broken backend: 1 (remove next)
  • healthy backend: 2 + (1 - utilization_ratio), range [2..3]

Among healthy backends, the most utilized backend gets the lowest score, making its copy the first candidate for removal -- freeing space where it is scarcest.

func (*OverReplicationCleaner) SetConfig

func (c *OverReplicationCleaner) SetConfig(cfg *config.ReplicationConfig)

SetConfig atomically stores the replication configuration.

type OverReplicationCleanerStore

OverReplicationCleanerStore is the narrow persistence surface the over-replication cleaner needs: over-replicated discovery + excess copy removal + per-backend quota stats for the deletion target. Declared locally so the worker does not pull in the full MetadataStore.

type OverReplicationCleanerStore interface {
    core.ReplicationStore
    core.QuotaStore
}

type PendingReaper

PendingReaper resolves abandoned PUT intents by inspecting the destination backend. The min-age window protects in-flight PUTs whose commit has not yet had a chance to clear the intent on the synchronous path.

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

func NewPendingReaper

func NewPendingReaper(deps PendingReaperDeps) *PendingReaper

NewPendingReaper creates a PendingReaper with the given dependencies.

func (*PendingReaper) ProcessPendingQueue

func (r *PendingReaper) ProcessPendingQueue(ctx context.Context) WorkSummary

ProcessPendingQueue runs one reaper tick: fetch stale intents, HEAD their destinations, and promote or drop based on what the backend reports. Returns the number of intents that completed (committed or dropped) and the number that failed (left for the next tick).

type PendingReaperDeps

PendingReaperDeps groups the pending reaper’s constructor parameters. Concurrency, MinAge, and BatchSize fall back to safe defaults when zero or negative.

type PendingReaperDeps struct {
    Ops         CleanupOps
    Placement   Placement
    Store       PendingReaperStore
    Concurrency int
    MinAge      time.Duration
    BatchSize   int
}

type PendingReaperStore

PendingReaperStore is the narrow persistence surface the pending reaper needs. Declared locally so the worker does not pull in the full MetadataStore.

type PendingReaperStore interface {
    core.PendingStore
}

type Placement

Placement is the store-coupled write-path facet (target selection, move, delete-or-enqueue) workers get from the BackendManager, kept apart from the runtime roles so the manager need not re-export runtime methods.

type Placement interface {
    SelectReplicaTarget(ctx context.Context, size int64, exclusion map[string]bool) (string, error)
    MoveObject(ctx context.Context, req *writepath.MoveRequest) (int64, error)
    DeleteOrEnqueue(ctx context.Context, be backend.ObjectBackend, backendName, key, reason string, sizeBytes int64)
}

type PlacementSet

PlacementSet maps an object key to the set of backends that already hold it. Built once from a copy map (key -> backend names) so planners can ask “is key already on backend?” without rescanning a slice or allocating a lookup map per object.

type PlacementSet map[string]map[string]struct{}

func NewPlacementSet

func NewPlacementSet(copyMap map[string][]string) PlacementSet

NewPlacementSet builds a PlacementSet from a key -> backend-names copy map.

func (PlacementSet) Has

func (p PlacementSet) Has(key, backend string) bool

Has reports whether key already lives on backend.

type RebalanceMove

RebalanceMove describes a single object move from one backend to another.

type RebalanceMove struct {
    ObjectKey   string
    FromBackend string
    ToBackend   string
    SizeBytes   int64
}

type Rebalancer

Rebalancer moves objects between backends to optimize space distribution.

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

func NewRebalancer

func NewRebalancer(ops Ops, placement Placement, store RebalancerStore) *Rebalancer

NewRebalancer creates a Rebalancer with fleet operations, write-path placement, and a metadata store.

func (*Rebalancer) Config

func (r *Rebalancer) Config() *config.RebalanceConfig

Config returns the current rebalance configuration.

func (*Rebalancer) ExecuteMoves

func (r *Rebalancer) ExecuteMoves(ctx context.Context, plan []RebalanceMove, strategy string, concurrency int) WorkSummary

ExecuteMoves runs the planned object moves with bounded concurrency. Skips individual moves that fail and continues with the rest, returning the count of successful moves.

func (*Rebalancer) ExecuteOneMove

func (r *Rebalancer) ExecuteOneMove(ctx context.Context, move RebalanceMove, strategy string) bool

ExecuteOneMove performs a single object move: stream the bytes from source to destination, swap the DB location with compare-and-swap, and delete the source copy. Returns true when all steps succeed.

func (*Rebalancer) PlanPackTight

func (r *Rebalancer) PlanPackTight(ctx context.Context, stats map[string]core.QuotaStat, batchSize int) ([]RebalanceMove, error)

PlanPackTight consolidates objects onto the most-utilized backends by pulling from the least-utilized. Sorts by percent full descending and only moves an object from a less-full source to a more-full destination. Skips moves that would not increase the destination’s packing ratio.

func (*Rebalancer) PlanSpreadEven

func (r *Rebalancer) PlanSpreadEven(ctx context.Context, stats map[string]core.QuotaStat, batchSize int) ([]RebalanceMove, error)

PlanSpreadEven equalizes utilization ratios across backends by moving objects from over-utilized backends to under-utilized ones. Returns nil when no backend has a usable byte limit.

func (*Rebalancer) Rebalance

func (r *Rebalancer) Rebalance(ctx context.Context, cfg config.RebalanceConfig) (WorkSummary, error)

Rebalance moves objects between backends to optimize space distribution. Returns the number of objects successfully moved.

func (*Rebalancer) SetConfig

func (r *Rebalancer) SetConfig(cfg *config.RebalanceConfig)

SetConfig atomically stores the rebalance configuration.

type RebalancerStore

RebalancerStore is the narrow persistence surface the rebalancer needs: per-backend listings + quota stats for the source/dest fill ratio. Declared locally so the worker does not pull in the full MetadataStore.

type RebalancerStore interface {
    core.ObjectStore
    core.QuotaStore
}

type ReconcileResult

ReconcileResult holds the outcome of a reconciliation pass for one backend.

type ReconcileResult struct {
    Imported        int `json:"imported"`
    Removed         int `json:"removed"`
    BackendsScanned int `json:"backends_scanned"`
}

type Reconciler

Reconciler scans backends for untracked objects and imports them into the metadata database.

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

func NewReconciler

func NewReconciler(syncer BackendSyncer, bucketNames []string) *Reconciler

NewReconciler creates a reconciler that uses the syncer’s SyncBackend to import untracked objects.

func (*Reconciler) Reconcile

func (r *Reconciler) Reconcile(ctx context.Context, backendName string) (*ReconcileResult, error)

Reconcile performs a full reconciliation for the given backend (or all backends if backendName is empty). Lists objects on each backend, diffs against DB entries, imports untracked objects, and removes stale entries.

func (*Reconciler) ReconcileStreaming

func (r *Reconciler) ReconcileStreaming(ctx context.Context, backendName string, observer progress.Observer) (*ReconcileResult, error)

ReconcileStreaming is Reconcile with a per-backend observer. onBackend, when non-nil, is called after each backend is reconciled so a streaming caller can report incremental progress; pass nil for the non-streaming path.

func (*Reconciler) Run

func (r *Reconciler) Run(ctx context.Context)

Run performs a full reconciliation pass: for each backend, list all objects and import any that are not tracked in the metadata database.

type RecorderProvider

RecorderProvider provides the shared per-backend accounting recorder. Prefer this over UsageAccessor for per-attempt API call / byte counters: APICall / Egress / Ingress / Operation document intent at the call site and prevent argument-order bugs.

type RecorderProvider interface {
    Acct() *accounting.Recorder
}

type ReplicationOutcome

ReplicationOutcome captures the per-object result of one ReplicateObject invocation. Counts are populated regardless of success so the reporter and unit tests can reason about retry behaviour without parsing log lines. NoTarget reflects whether the loop exited because target selection ran out, not whether any individual attempt failed.

type ReplicationOutcome struct {
    Key          string // object key
    Created      int    // copies successfully recorded
    CopyErrors   int    // source -> target stream copies that errored
    RecordErrors int    // RecordReplica failures (after the copy succeeded)
    Superseded   int    // RecordReplica returned inserted=false (source row gone)
    NoTarget     bool   // selection failed before `needed` was reached
}

func (ReplicationOutcome) Failed

func (o ReplicationOutcome) Failed() int

Failed reports the total number of failed copy attempts for this object, irrespective of failure stage.

type Replicator

Replicator creates additional copies of under-replicated objects across backends.

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

func NewReplicator

func NewReplicator(ops Ops, placement Placement, store ReplicatorStore) *Replicator

NewReplicator creates a Replicator with fleet operations, write-path placement, and a metadata store.

func (*Replicator) CleanupOrphan

func (r *Replicator) CleanupOrphan(ctx context.Context, backendName, key string, sizeBytes int64)

CleanupOrphan deletes an object from a backend when the DB record was not created (e.g. source was deleted during replication). Looks up the backend by name and dispatches to DeleteOrEnqueue, which handles its own API accounting and orphan-byte tracking.

func (*Replicator) Config

func (r *Replicator) Config() *config.ReplicationConfig

Config returns the current replication configuration.

func (*Replicator) CopyToReplica

func (r *Replicator) CopyToReplica(ctx context.Context, key string, copies []core.ObjectLocation, target string) (string, int64, error)

copyToReplica reads the object from an existing copy and writes it to the target backend. Tries each existing copy in order for failover. Returns the source backend name that was successfully read from and the size_bytes recorded on that source’s ObjectLocation row (the size of the bytes that were actually transferred). The input slice is cloned before sorting so callers retain their original ordering — without the clone, sort.Slice reorders the caller’s slice in place and the caller’s later reads see a different element at each index.

func (*Replicator) FindReplicaTarget

func (r *Replicator) FindReplicaTarget(ctx context.Context, key string, size int64, exclusion map[string]bool) string

FindReplicaTarget selects a backend for a replication copy using the same routing strategy as normal writes. Returns empty string if no suitable target exists.

func (*Replicator) IsBackendHealthy

func (r *Replicator) IsBackendHealthy(name string) bool

IsBackendHealthy returns true if the backend has a closed circuit breaker or has no circuit breaker wrapper.

func (*Replicator) Replicate

func (r *Replicator) Replicate(ctx context.Context, cfg config.ReplicationConfig, observer progress.Observer) (int, error)

Replicate finds under-replicated objects and creates additional copies to reach the target replication factor. Returns the number of copies created. observer, when non-nil, receives a start and end step per object replicated.

func (*Replicator) ReplicateObject

func (r *Replicator) ReplicateObject(ctx context.Context, key string, existingCopies []core.ObjectLocation, needed int) ReplicationOutcome

ReplicateObject creates up to `needed` additional copies of a single object. Returns a ReplicationOutcome the caller can use to drive metrics, audit, and log reporting without re-parsing logs.

Per-attempt diagnostic logs are preserved so incident responders can still trace each failed retry; the outcome is a *structured summary* on top of those, not a replacement.

func (*Replicator) SetConfig

func (r *Replicator) SetConfig(cfg *config.ReplicationConfig)

SetConfig atomically stores the replication configuration.

func (*Replicator) UnhealthyBackends

func (r *Replicator) UnhealthyBackends(threshold time.Duration) []string

UnhealthyBackends returns backend names whose circuit breakers have been open longer than the given threshold. Returns nil when all backends are healthy or circuit breakers are not enabled.

type ReplicatorStore

ReplicatorStore is the narrow persistence surface the replicator needs: under-replicated discovery + per-replica record / location updates. Declared locally so the worker does not pull in the full MetadataStore.

type ReplicatorStore interface {
    core.ObjectStore
    core.ReplicationStore
}

type Scrubber

Scrubber periodically verifies stored object integrity by reading objects from backends, computing their SHA-256 hash, and comparing against the stored content hash. Also supports backfilling hashes for objects that were written before integrity was enabled.

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

func NewScrubber

func NewScrubber(deps ScrubberDeps) *Scrubber

NewScrubber creates a Scrubber with the given dependencies.

func (*Scrubber) Backfill

func (s *Scrubber) Backfill(ctx context.Context, batchSize, offset int, observer progress.Observer) (WorkSummary, int)

Backfill reads objects that have no stored content hash, computes the SHA-256 digest, and stores it in the database. Processes up to batchSize objects starting at the given offset. observer, when non-nil, receives a start step before each object is hashed and an end step after, carrying the per-object outcome and duration. Returns the cycle summary and the next offset for pagination (0 when done).

func (*Scrubber) Config

func (s *Scrubber) Config() *config.IntegrityConfig

Config returns the current integrity configuration.

func (*Scrubber) Scrub

func (s *Scrubber) Scrub(ctx context.Context, batchSize int, observer progress.Observer) WorkSummary

Scrub verifies a batch of objects with stored content hashes. Returns the number of objects checked and the number of hash mismatches found.

func (*Scrubber) SetConfig

func (s *Scrubber) SetConfig(cfg *config.IntegrityConfig)

SetConfig atomically stores the integrity configuration.

type ScrubberDeps

ScrubberDeps groups the scrubber’s constructor dependencies. Encryptor is optional (nil when encryption is disabled).

type ScrubberDeps struct {
    Ops       ScrubberOps
    Placement Placement
    Store     ScrubberStore
    Encryptor *encryption.Encryptor
}

type ScrubberOps

ScrubberOps is the dependency contract for Scrubber. It omits AdmissionControl and BackendAccess because integrity checks are best-effort background work.

type ScrubberOps interface {
    DataMover
    UsageAccessor
    RecorderProvider
}

type ScrubberStore

ScrubberStore is the narrow persistence surface the scrubber needs: integrity row reads/writes. Declared locally so the worker does not pull in the full MetadataStore.

type ScrubberStore interface {
    core.IntegrityStore
}

type SourceCandidates

SourceCandidates bundles a source backend’s candidate objects with their placement set, fetched together so a planner pays the two store queries once per source.

type SourceCandidates struct {
    Objects   []core.ObjectLocation
    Placement PlacementSet
}

type UsageAccessor

UsageAccessor provides usage tracking.

type UsageAccessor interface {
    Usage() *counter.UsageTracker
}

type WorkSummary

WorkSummary is the uniform result of one worker cycle. It replaces the ad-hoc count tuples (checked/failed, processed/failed, moved) each worker used to return, so partial-failure reporting and metric labels are consistent.

type WorkSummary struct {
    Planned   int           // items the cycle set out to process
    Attempted int           // items the per-item function ran (succeeded + failed)
    Succeeded int           // items that completed successfully
    Failed    int           // items attempted but not completed
    Skipped   int           // items declined before any work
    Duration  time.Duration // wall-clock time for the cycle
}

func (WorkSummary) Outcome

func (s WorkSummary) Outcome() string

Outcome classifies the cycle for metric labels: success (work done, no failures), partial (some succeeded, some failed), failed (only failures), or empty (nothing succeeded or failed).

Generated by gomarkdoc