s3-orchestrator

proxy

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

Package proxy is the domain orchestration layer that coordinates multi-backend S3 storage. It routes writes, manages failover reads, handles multipart uploads, drains backends, and exposes dashboard data. Workers receive the Ops interface instead of direct access.

Index

type BackendManager

BackendManager manages multiple storage backends with quota tracking. It holds the backend runtime (non-store infrastructure: backends, usage, admission, draining, metrics) as a named field reached via Runtime(), plus the per-role store views and hot-reloadable config. Store-touching write-path helpers are methods on *BackendManager (manager_writepath.go); pure infra primitives stay on the runtime.

Workers (rebalancer, replicator, scrubber, …) are resolved through DI at the call site rather than carried on the manager.

The drain manager is an injected collaborator. It is nil-able; the methods that consult it (FlushUsage, ClearDrainState, GetDashboardData) nil-guard the field so a manager built without drain stays usable.

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

func NewBackendManager

func NewBackendManager(cfg *BackendManagerConfig) *BackendManager

NewBackendManager constructs a BackendManager. Required dependencies (cfg, Stores, Dashboard, Metrics) panic via must.NotNil at construction so a wiring bug surfaces immediately at DI assembly rather than NPE’ing N call frames deep on the first request. Numeric config invariants (negative timeouts, ordering rules) are the config validator’s responsibility; the constructor trusts the values it receives.

func (*BackendManager) AdmissionSem

func (m *BackendManager) AdmissionSem() chan struct{}

AdmissionSem returns the shared admission semaphore, or nil if none is configured. The HTTP admission controller should use this channel so that HTTP requests and background services share one concurrency budget.

func (*BackendManager) BackendOrder

func (m *BackendManager) BackendOrder() []string

BackendOrder forwards to the runtime. The reconciler iterates the fleet in this order while reconciling backend state against the stores.

func (*BackendManager) ClearCache

func (m *BackendManager) ClearCache()

ClearCache removes all entries from the location cache.

func (*BackendManager) ClearDrainState

func (m *BackendManager) ClearDrainState()

ClearDrainState removes all entries from the draining map. Used by tests to reset state between runs. No-op when the manager has no drain manager.

func (*BackendManager) Close

func (m *BackendManager) Close()

Close stops every background cache eviction goroutine the manager owns: the object location cache and the multipart per-upload DEK cache. Safe to call multiple times.

func (*BackendManager) CountActiveMultipartUploads

func (m *BackendManager) CountActiveMultipartUploads(ctx context.Context, bucketPrefix string) (int64, error)

CountActiveMultipartUploads delegates to the multipart store. Exposed for the s3api bucket-delete pre-check so the transport layer does not need to reach into the persistence layer directly.

func (*BackendManager) DeleteOrEnqueue

func (m *BackendManager) DeleteOrEnqueue(ctx context.Context, be backend.ObjectBackend, backendName, key, reason string, sizeBytes int64)

DeleteOrEnqueue forwards to the write coordinator. The worker Placement and drain Mover interfaces call it on *BackendManager.

func (*BackendManager) Drain

func (m *BackendManager) Drain() *drain.Manager

Drain returns the drain manager, or nil when the manager was built without one. Callers that touch the result must nil-guard.

func (*BackendManager) FlushUsage

func (m *BackendManager) FlushUsage(ctx context.Context) error

FlushUsage flushes accumulated in-memory usage counters to the database. Backends that have completed draining are skipped because their DB records (including backend_usage) have been removed. When DrainManager has not been wired (tests that do not exercise drain behavior) the skip set is empty and every backend’s counters flush.

func (*BackendManager) GetDashboardData

func (m *BackendManager) GetDashboardData(ctx context.Context) (*dashboard.Data, error)

GetDashboardData delegates to the dashboard.Aggregator and enriches the result with drain status and circuit-breaker health from the BackendManager’s in-memory state.

func (*BackendManager) GetDirectoryChildren

func (m *BackendManager) GetDirectoryChildren(ctx context.Context, prefix, startAfter string, maxKeys int) (*core.DirectoryListResult, error)

GetDirectoryChildren delegates to the dashboard.Aggregator.

func (*BackendManager) IntegrityConfig

func (m *BackendManager) IntegrityConfig() *config.IntegrityConfig

IntegrityConfig returns the current integrity configuration.

func (*BackendManager) LifecycleConfig

func (m *BackendManager) LifecycleConfig() *config.LifecycleConfig

LifecycleConfig returns the current lifecycle configuration.

func (*BackendManager) MoveObject

func (m *BackendManager) MoveObject(ctx context.Context, req *writepath.MoveRequest) (int64, error)

MoveObject forwards to the write coordinator’s shared move primitive so the StreamCopy + MoveObjectLocation CAS + orphan-cleanup + source-delete accounting all funnel through one implementation.

func (*BackendManager) Multipart

func (m *BackendManager) Multipart() *multipart.Manager

Multipart returns the multipart upload lifecycle manager. Exposed so transport and DI callers can reach multipart functionality without touching the unexported field directly.

func (*BackendManager) NearUsageLimit

func (m *BackendManager) NearUsageLimit(threshold float64) bool

NearUsageLimit returns true if any backend is approaching its usage limits.

func (*BackendManager) Objects

func (m *BackendManager) Objects() *object.Manager

Objects returns the object CRUD manager. Same accessor rationale as Multipart().

func (*BackendManager) ProcessLifecycleRules

func (m *BackendManager) ProcessLifecycleRules(ctx context.Context, rules []config.LifecycleRule) (deleted, failed int)

ProcessLifecycleRules evaluates all lifecycle rules and deletes expired objects. Returns total deleted and failed counts. Terminates processing of a rule when a full batch produces zero successful deletions, preventing infinite loops when backends are unhealthy.

func (*BackendManager) ReconcileBackend

func (m *BackendManager) ReconcileBackend(ctx context.Context, backendName, bucket string, knownBuckets []string) (*worker.ReconcileResult, error)

ReconcileBackend reconciles a single backend against the metadata store using a bounded-memory sorted-merge: both sides are walked in lex key order and diffed in lockstep. The S3 walk and DB cursor each cap their in-flight buffer, so memory is independent of object count.

Behaviour: imports keys present on the backend but not in the DB, and deletes DB rows whose keys are no longer on the backend. Keys owned by sibling virtual buckets stored on the same backend are left alone in both directions - sibling buckets are reconciled by their own pass.

func (*BackendManager) ReconcileUsage

func (m *BackendManager) ReconcileUsage(ctx context.Context) (map[string]int64, error)

ReconcileUsage recomputes each backend’s bytes_used counter from the object ledger, correcting drift in the incrementally maintained counter. Part of the BackendSyncer contract the reconciler drives every pass; also exposed to the admin reconcile-usage endpoint.

func (*BackendManager) RecordUsage

func (m *BackendManager) RecordUsage(backendName string, apiCalls, egress, ingress int64)

RecordUsage increments the in-memory usage counters for a backend. Exposed for admin operations that bypass the normal manager request path.

func (*BackendManager) RedisCounterConfigured

func (m *BackendManager) RedisCounterConfigured() bool

RedisCounterConfigured returns true when the counter backend is a Redis backend, regardless of health status. Used by the flush service to decide whether an advisory lock is needed - the lock must be held even during fallback to prevent double-counting when Redis recovers mid-flush.

func (*BackendManager) Runtime

func (m *BackendManager) Runtime() *infra.BackendRuntime

Runtime returns the backend runtime so workers, drain, and transport can depend on it directly for fleet/admission/usage primitives.

func (*BackendManager) SelectReplicaTarget

func (m *BackendManager) SelectReplicaTarget(ctx context.Context, size int64, exclusion map[string]bool) (string, error)

SelectReplicaTarget picks a target backend for a replication copy using the same routing strategy as normal writes. Excludes backends that already hold a copy of the object.

func (*BackendManager) SetIntegrityConfig

func (m *BackendManager) SetIntegrityConfig(cfg *config.IntegrityConfig)

SetIntegrityConfig atomically stores the integrity configuration. The scrubber’s own SetConfig is invoked separately by the caller (serve) because the scrubber is resolved through DI rather than held on the manager.

func (*BackendManager) SetLifecycleConfig

func (m *BackendManager) SetLifecycleConfig(cfg *config.LifecycleConfig)

SetLifecycleConfig atomically stores the lifecycle configuration.

func (*BackendManager) SetUsageFlushConfig

func (m *BackendManager) SetUsageFlushConfig(cfg *config.UsageFlushConfig)

SetUsageFlushConfig atomically stores the usage flush configuration.

func (*BackendManager) SyncBackend

func (m *BackendManager) SyncBackend(ctx context.Context, backendName, bucket string, knownBuckets []string) (imported, skipped int, err error)

SyncBackend scans a backend’s S3 bucket and imports pre-existing objects into the proxy database. Objects already tracked for the backend are skipped. knownBuckets is the full list of configured virtual bucket names, used to distinguish objects belonging to other buckets from externally-uploaded objects that need the bucket prefix prepended. Returns counts of imported vs skipped objects.

func (*BackendManager) UpdateQuotaMetrics

func (m *BackendManager) UpdateQuotaMetrics(ctx context.Context) error

UpdateQuotaMetrics forwards to the runtime. The usage-flush and reconcile services consume it alongside the manager’s store-coupled helpers, so the manager exposes it as part of its orchestration surface.

func (*BackendManager) UpdateUsageLimits

func (m *BackendManager) UpdateUsageLimits(limits map[string]core.UsageLimits)

UpdateUsageLimits replaces the per-backend usage limits. Safe to call concurrently with request handling.

func (*BackendManager) UsageFlushConfig

func (m *BackendManager) UsageFlushConfig() *config.UsageFlushConfig

UsageFlushConfig returns the current usage flush configuration.

type BackendManagerConfig

BackendManagerConfig groups the constructor parameters by capability so contributors can see at a glance which fields belong together: topology, persistence, runtime policy, optional features, operational deps, and pre-built collaborators. Each sub-struct documents its own field semantics.

type BackendManagerConfig struct {
    Runtime       *infra.BackendRuntime // backend fleet/admission/usage/metrics infrastructure, built by the composition root
    Storage       StorageDeps
    Stores        StoreDeps
    Policies      PolicyConfig
    Features      FeatureDeps
    Operations    OperationalDeps
    Collaborators Collaborators
}

type Collaborators

Collaborators groups the sub-managers built by the composition root and injected so the drain manager (which needs the write coordinator as its mover and the multipart abort hook) and the BackendManager share the same instances.

Coord, Multipart, and IntegrityCfg are required: the drain manager and the BackendManager must hold the same coordinator, multipart manager, and integrity-config pointer. Drain is nil-able; the methods that consult it (FlushUsage, ClearDrainState, GetDashboardData) nil-guard the field.

type Collaborators struct {
    Coord        *writepath.Coordinator
    Multipart    *multipart.Manager
    Drain        *drain.Manager
    IntegrityCfg *syncutil.AtomicConfig[config.IntegrityConfig]
}

type FeatureDeps

FeatureDeps groups optional capabilities. Each field is nil-able and disables the corresponding feature when left zero.

type FeatureDeps struct {
    Encryptor      *encryption.Encryptor  // nil when encryption is disabled
    CounterBackend counter.CounterBackend // nil uses LocalCounterBackend
    ObjectCache    objcache.ObjectCache   // nil when object data caching is disabled
}

type ManagerStores

ManagerStores is the narrow persistence surface BackendManager itself touches: object import / delete, cleanup-queue sweep, lifecycle expiry listing, usage-delta flush, and the multipart count it exposes to the s3api transport. Sub-managers (object, writepath, multipart, readpath) receive their own narrower role-composite interfaces through their constructors; the *core.MetadataStore handed in via BackendManagerConfig is the composition-root concrete that satisfies all of them.

type ManagerStores interface {
    core.ObjectStore
    core.CleanupStore
    core.ExpiredObjectsLister
    core.UsageFlusher
    core.MultipartStore
    core.QuotaStore
}

type OperationalDeps

OperationalDeps groups telemetry, concurrency, and observability callbacks the manager exposes to operators and to long-running background services.

type OperationalDeps struct {
    Metrics metrics.Deps
    // AdmissionSem is the shared concurrency semaphore for write-class
    // traffic. In split mode (MaxConcurrentReads + MaxConcurrentWrites)
    // it is sized to MaxConcurrentWrites and is shared between HTTP
    // writes and all background workers; reads run on a separate sem
    // created in transport/httpserver/routes.go. In merged mode
    // (MaxConcurrentRequests only) it is the global pool for every HTTP
    // request and every background worker. nil disables admission entirely
    // (no cap installed). See admissionSemFor in internal/di/backend.go
    // for the sizing rules.
    AdmissionSem chan struct{}
    // ReplicationFactor is invoked by the metrics collector when refreshing
    // the under-replicated-objects gauge. Returns 0 when replication is
    // disabled. Lazy-evaluated so it can resolve the live replicator's
    // configured factor (which is hot-reloadable).
    ReplicationFactor func() int
}

type PolicyConfig

PolicyConfig groups runtime tunables that shape how the manager behaves across normal and degraded operation. None of these enable a feature; they configure existing behavior.

type PolicyConfig struct {
    BackendTimeout time.Duration
    CacheTTL       time.Duration
    UsageLimits    map[string]core.UsageLimits
    // RoutingStrategy selects write-target ordering: pack vs spread.
    RoutingStrategy config.RoutingStrategy
    // ParallelBroadcast fans out reads in parallel during degraded mode.
    ParallelBroadcast bool
    // DegradedBroadcastParallelism caps concurrent probes during a
    // parallel degraded-mode broadcast. 0 = no cap (every backend
    // probed at once, the historical behaviour).
    DegradedBroadcastParallelism int
    // DisableDegradedReads opts the read path out of broadcasting on DB outage.
    DisableDegradedReads bool
    // PendingEnabled toggles the PUT-before-COMMIT pending-row pattern
    // (write_path.pending_pattern.enabled). When false the manager skips
    // pending-intent inserts and pending-promotion paths and falls back
    // to the legacy cleanup-on-failure flow.
    PendingEnabled bool
    // MaxObjectSizes is the per-backend max object size in bytes (0 = unlimited).
    MaxObjectSizes map[string]int64
}

type StorageDeps

StorageDeps groups the backend-fleet topology: the set of object backends to route across and the deterministic per-strategy iteration order.

type StorageDeps struct {
    Backends map[string]backend.ObjectBackend
    Order    []string
}

type StoreDeps

StoreDeps groups the persistence dependencies. Metadata stays as the wide core.MetadataStore because BackendManager is the proxy subtree’s composition root — it routes the concrete store into the narrow interfaces each sub-manager declares. Dashboard is already narrow.

type StoreDeps struct {
    Metadata  core.MetadataStore
    Dashboard core.DashboardStore
}

Generated by gomarkdoc