ecluse:ecluse-core
Safe HaskellNone
LanguageGHC2021

Ecluse.Core.Worker

Description

The mirror worker: the supervised consume loop that turns enqueued jobs into mirrored packages.

The worker is the consumer end of the demand-driven mirror queue (see Ecluse.Core.Queue). The consume loop long-polls the queue, resolves each received job's ecosystem bundle (WorkerPolicy, keyed by the job's own ecosystem; a job whose ecosystem carries none is fail-closed), and through that bundle:

  1. probes the mirror target for the job's version, acking a confirmed-present duplicate outright (demand-driven enqueue means a fleet-wide install of a novel version enqueues many jobs for it; only the first has work to do),
  2. re-evaluates current policy for the version through the same rules and single-version fetch the serve path gates with, so a version denied since its serve-time admit is dropped rather than mirrored,
  3. fetches the artifact bytes from the public upstream named on the job,
  4. verifies those bytes against the integrity digests of the artifact the re-evaluation re-admitted (the floor-checked, current-metadata set; the queue payload carries no digest at all),
  5. assembles the ecosystem's publish document from the re-admitted artifact's descriptor and publishes it to the mirror target (the bundle's married publish capability, resolved at the composition root with the bearer from the Ecluse.Core.Credential provider), and
  6. acknowledges the job.

See individual modules for detailed behaviour: * Ecluse.Core.Worker.Integrity for the security gate on artifact digests. * Ecluse.Core.Worker.Loop for supervision and graceful shutdown. * Ecluse.Core.Worker.Job for ack semantics within the visibility budget.

See docs/architecture/cloud-backends.md → "Mirror Queue" and "Process model".

Synopsis

Worker runtime

data WorkerRuntime Source #

The runtime backends the mirror worker is closed over: exactly the effectful capabilities the consume loop needs to poll, fetch, verify, publish, and record. A record of concrete handles and abstract ports (the Handle pattern), assembled by the composition root (workerRuntimeOf) and read by the loop through the WorkerM reader.

The mirror queue is the demand-driven hand-off the loop consumes; the untrusted data-plane manager fetches the artifact bytes (the validating TLS manager, over an https-only dist.tarball); the heartbeat is the loop's liveness surface. The mirror write is not a runtime slot: it rides each ecosystem's bundle (wpPublish), so every job publishes through its own ecosystem's married capability. The metric and tracing ports are the abstract recording interfaces (Ecluse.Core.Telemetry.Record, Ecluse.Core.Telemetry.Span); the application supplies their OpenTelemetry-backed implementations, so the loop records without naming a telemetry backend. There is no log field: the loop logs through the ambient katip context the entry point establishes.

Constructors

WorkerRuntime 

Fields

  • wrQueue :: MirrorQueue

    The mirror-queue handle the consume loop long-polls and acks against.

  • wrManager :: Manager

    The validating-TLS data-plane manager for the untrusted artifact fetch (over an https-only dist.tarball).

  • wrHeartbeat :: WorkerHeartbeat

    The consume-loop heartbeat, advanced on every successful poll and every completed job (see recordWorkerProgress) and read by the liveness probe.

  • wrMetrics :: WorkerMetricsPort

    The metric-recording port the worker emits its ecluse.mirror.* job signals through.

  • wrTracing :: WorkerTracingPort

    The tracing port the worker opens its per-job span through.

  • wrInjectTraceContext :: forall (m :: Type -> Type) a. (KatipContext m, MonadIO m) => m a -> m a

    Evaluate and inject the current OpenTelemetry correlation payload into the katip context for the inner action.

  • wrPolicies :: WorkerPolicies

    The per-ecosystem re-evaluation bundles, keyed by a job's ecosystem. The worker re-runs current policy against a job's version before it mirrors it, so a policy that has tightened toward deny since the job was enqueued drops the job rather than freezing a now-disallowed version into the trusted mirror store.

Instances

Instances details
MonadReader WorkerRuntime WorkerM Source # 
Instance details

Defined in Ecluse.Core.Worker.Types

Per-ecosystem ingest re-evaluation

data WorkerPolicy Source #

The per-ecosystem bundle the worker dispatches every job through: a resolver that fetches and projects the single version's metadata, the prepared rule set, the integrity floor, the tarball-host gate, the artifact request formation, the married mirror-write capability, and the wall-clock the age rules read.

The resolver is the shared single-version fetch-and-project (fetchVersionDetails over the guarded public origin, wired by the composition root); the rules are the same prepared rules the serve path gates with; the floor and host gate are the mount's own configured policy values; and the request formation is the mount ecosystem's own (pdBuildArtifactRequestByUrl) -- so the worker's ingest decision and the serve-time decision run one codepath (admitArtifact) over one policy, and any per-source breaker state is shared, never forked. The publish capability is likewise the mount's own, so the presence probe and the mirror write speak the job ecosystem's protocol at that ecosystem's declared mirror target, never a neighbour's.

Constructors

WorkerPolicy 

Fields

  • wpResolveVersion :: PackageName -> Version -> IO VersionEvaluation

    Resolve and project one version's metadata through the guarded public origin, classifying the outcome (fetchVersionDetails). Total by type: the fetch reports every failure -- transport included -- in its typed channel, and each classifies as a VersionMetadataUnavailable value.

  • wpRules :: [PreparedRule]

    The prepared rule set evaluated against the resolved version under current policy (the same rules the serve path gates the public version set with).

  • wpMinIntegrity :: MinIntegrity

    The mount's own public-integrity floor (pdMinIntegrity), re-applied at ingest through the shared admission gate.

  • wpArtifactHostHonoured :: Maybe HostPort -> Bool

    The mount's own tarball-host gate (tarballHostHonoured, closed against the public upstream authority), re-checked on the extracted host:port of the job's fetch URL (Nothing, an unextractable authority, is refused): the queue payload is a trust boundary.

  • wpBuildArtifactRequest :: Limits -> Manager -> Text -> Maybe Secret -> Text -> Either UrlFormationError Request

    Form the artifact GET request for a job's authoritative artifact URL: the mount ecosystem's own request formation (pdBuildArtifactRequestByUrl), so a job's bytes are fetched with the same request formation the serve path streams with. Riding this bundle means a job whose ecosystem has none never reaches a fetch: it is fail-closed with the rest of the bundle.

  • wpPublish :: MirrorPublish

    The mount's married mirror-write capability (newMirrorPublish: the adapter's protocol codec over the shared publish transport, bound to the mount's declared mirror target). The presence probe and the verified-bytes publish both ride it, so a job can only ever consult the capability keyed by its own ecosystem; a job whose ecosystem carries no bundle is fail-closed before any of this runs.

  • wpArtifactLimits :: Limits

    The bounded-fetch budget for the artifact download (fetchArtifactBytes): the composition root sets its maxBodyBytes from the memory plan's mirror-artifact tenant (in Ecluse.Composition.MemoryPlan), so the worker never buffers a tarball whose transient publish envelope would breach the heap ceiling the plan partitions.

  • wpNow :: IO UTCTime

    The wall-clock "now" for the rules' EvalContext; injected so the time-sensitive age gate is deterministic under test.

type WorkerPolicies = Map Ecosystem WorkerPolicy Source #

The worker's per-ecosystem re-evaluation bundles, keyed by the ecosystem a job's package belongs to (pkgEcosystem). Built once at boot and shared with the serve mounts; a job whose ecosystem is absent here is fail-closed (dropped), never mirrored unvetted.

The worker monad

data WorkerM a Source #

The mirror worker's monad: a reader over the WorkerRuntime layered on katip's logging context.

A newtype over ReaderT WorkerRuntime (KatipContextT IO) so its instances are this module's to control and call sites name one concrete monad. The derived instances give reader access to the runtime (MonadReader WorkerRuntime), arbitrary effects (MonadIO), the unlift capability (MonadUnliftIO) the loop's tryAny and the per-job span bracket need, and the katip classes (Katip, KatipContext) so a structured log call composes through the ambient context the entry point establishes.

The katip base is a reader, never a StateT, so the logging context behaves correctly across the loop (see docs/architecture/technology-stack.md → "Key Decisions").

Instances

Instances details
MonadIO WorkerM Source # 
Instance details

Defined in Ecluse.Core.Worker.Types

Methods

liftIO :: IO a -> WorkerM a #

Applicative WorkerM Source # 
Instance details

Defined in Ecluse.Core.Worker.Types

Methods

pure :: a -> WorkerM a #

(<*>) :: WorkerM (a -> b) -> WorkerM a -> WorkerM b #

liftA2 :: (a -> b -> c) -> WorkerM a -> WorkerM b -> WorkerM c #

(*>) :: WorkerM a -> WorkerM b -> WorkerM b #

(<*) :: WorkerM a -> WorkerM b -> WorkerM a #

Functor WorkerM Source # 
Instance details

Defined in Ecluse.Core.Worker.Types

Methods

fmap :: (a -> b) -> WorkerM a -> WorkerM b #

(<$) :: a -> WorkerM b -> WorkerM a #

Monad WorkerM Source # 
Instance details

Defined in Ecluse.Core.Worker.Types

Methods

(>>=) :: WorkerM a -> (a -> WorkerM b) -> WorkerM b #

(>>) :: WorkerM a -> WorkerM b -> WorkerM b #

return :: a -> WorkerM a #

Katip WorkerM Source # 
Instance details

Defined in Ecluse.Core.Worker.Types

KatipContext WorkerM Source # 
Instance details

Defined in Ecluse.Core.Worker.Types

MonadUnliftIO WorkerM Source # 
Instance details

Defined in Ecluse.Core.Worker.Types

Methods

withRunInIO :: ((forall a. WorkerM a -> IO a) -> IO b) -> WorkerM b Source #

MonadReader WorkerRuntime WorkerM Source # 
Instance details

Defined in Ecluse.Core.Worker.Types

runWorkerM :: LogEnv -> SimpleLogPayload -> WorkerRuntime -> WorkerM a -> IO a Source #

Run a WorkerM against the WorkerRuntime and the katip logging environment and initial context the entry point supplies, yielding the underlying IO action. This is the boundary where the worker's WorkerM code is discharged to IO.

The LogEnv (the structured-log scribes) and the initial context payload are passed in rather than read from the runtime, so the application owns the log stream and the trace-correlation dd enrichment: it resolves the dd identity and hands it here as the initial context, so every line the loop emits carries dd. The loop narrows the namespace with katip's combinators on top as it logs.

Loop and job processing (exposed for direct testing)

workerLoop :: SupervisionPolicy -> WorkerM Void Source #

The continuous consume loop: long-poll for a batch, process it, repeat, supervised under the given policy (the composition root names the wiring faults that must fail up rather than retry; tests inject their own).

A failed poll arrives as the handle's typed QueueFault value: it is logged and the step backs off and polls again, so a queue outage cannot kill the worker thread. A successful poll advances the heartbeat (whether or not the batch was empty), and processBatch advances it again after each completed job, so a liveness probe sees the loop is alive even while a healthy worker grinds through a long batch of large artifacts; an idle queue is a healthy empty poll, not a stall. The heartbeat advances only on demonstrated progress (a successful receive or a completed job), so a worker that cannot poll at all (a persistently faulting receive) keeps retrying but never advances it: the heartbeat goes stale and /livez fails, surfacing a fully-dead worker for the orchestrator to restart.

processBatch :: [QueueMessage] -> WorkerM () Source #

Process one received batch sequentially, so each job gets the full visibility budget rather than competing with its batch-mates for it. A batch is at most the queue's configured batch size (≤ 10), so sequential processing is a deliberate throughput-vs-budget choice, not a scaling bottleneck.

The liveness heartbeat advances after each completed job, not once for the whole batch, so the /livez staleness bound (workerHeartbeatStaleAfter) need only cover one job's worst case rather than a whole sequential batch of large-artifact publishes: a healthy worker mid-batch is not mistaken for a stalled one.

processJob :: ReceiptHandle -> MirrorJob -> WorkerM JobOutcome Source #

Process one mirror job end to end: probe the mirror target for the job's version (a confirmed-present version is acked outright, the duplicate-suppression short-circuit), then re-evaluate current policy, and only on a current admit fetch the artifact, verify it against the integrity digests of the artifact that re-evaluation re-admitted, and publish it to the mirror target. Returns the JobOutcome that decides ack vs. redeliver.

The presence probe exists for the enqueue-to-availability window: mirroring is demand-driven, so every public-leg admit of a still-unmirrored version enqueues its own job, and a fleet-wide install of a novel version enqueues many. Without the probe each duplicate pays a full artifact download and an integrity recompute before the publish discovers the version is already present (the idempotent already-present answer); with it, a duplicate costs one metadata round trip. The probe is an optimisation, never a gate: it skips only work whose publish would have been that no-op, so the policy re-evaluation below still guards every artifact that actually publishes.

The policy re-evaluation is the ingest-time gate. The version was gated at serve time, but the enqueue-to-process window is asynchronous and unbounded, so policy may have tightened toward deny since (a new denylist entry, a freshly-published advisory, a rule-config change). The worker re-runs the same rules the serve path gates with, over the version resolved through the same single-version fetch-and-project, so a now-denied version is dropped (acked, never published) rather than frozen into the rule-exempt trusted mirror store; a version the upstream has since withdrawn is likewise dropped, while metadata that cannot be re-fetched (or a rule that cannot be computed) leaves the job for redelivery. A current admit carries the re-admitted artifact's integrity digests to the tamper gate, so the fetched bytes are verified against the exact set the integrity floor cleared (the queue payload carries no digest at all): a tampered or corrupt artifact fails the job with no publish, since the mirror is later served without the rules.

The receipt handle is taken so a long publish can extendVisibility to hold the message before its window lapses.

The per-job domain span (the worker tracing port) wraps the whole probe → re-evaluate → fetch → verify → publish, projecting the terminal outcome onto the span so a refused or dropped job is explainable from the trace, and linking back to the request that enqueued the job through the trace context the job carries (jobTraceContext). The span body is discharged to IO through the unlift, so the loop's structured log lines still compose through the ambient katip context.

data JobOutcome Source #

The terminal outcome of processing one mirror job, deciding whether the message is acked or left to redeliver.

Constructors

Succeeded

The publish succeeded, so the job is acked. This covers an idempotent redelivery too: a version already present at the mirror target answers a status the ecosystem's codec classifies as success (npm's 409), so it surfaces here as Succeeded rather than a distinct case -- as does the same presence confirmed by the pre-fetch probe, before any bytes moved.

Dropped Text

A non-retryable rejection: the bytes did not match the re-admitted artifact's digest (tamper), or the publish URL was unformable (misconfiguration). Redelivery cannot help, so the job is acked to retire it after alarming. Carries the reason.

DeadLettered Text

A terminal fault the backend dead-letters: an artifact past the plan-sized byte cap can never succeed and re-fetches identical over-cap bytes on every redelivery, so it is not acked (a plain delete would silently discard it on a durable queue) but handed to the queue's deadLetter terminus -- the in-memory backend drops it, a durable queue rides it to the dead-letter queue for forensic retention. Carries the reason.

Retried Text

A transient fault: a fetch failure, or a registry rejection worth retrying. The message is left un-acked so it redelivers. Carries the reason.

Instances

Instances details
Show JobOutcome Source # 
Instance details

Defined in Ecluse.Core.Worker.Job

Eq JobOutcome Source # 
Instance details

Defined in Ecluse.Core.Worker.Job

workerPublishVisibilityBudget :: Seconds Source #

The visibility window one publish is given before its message could redeliver mid-write. Sized to comfortably cover a publish of the largest artifact the memory plan's fetch cap admits (the mirror-artifact tenant, at most 512 MiB at its ceiling): even over a slow mirror-target link (a conservative ~2 MiB/s floor) that uploads in well under this, so a successful publish never redelivers mid-flight. A failed publish does not wait this out -- the failure path resets the message to visible at once (see releaseForRetry) -- so the generous hold costs nothing on the retry path; this is the background worker's correct trade (never interrupt a slow success; retry latency on failure does not matter).

The liveness staleness bound (workerHeartbeatStaleAfter) is sized to exceed a fetch and a publish of this budget, so a healthy worker mid-publish is never read as stalled; Ecluse.Worker.LivenessSpec pins that relationship so the two constants cannot drift apart.

Liveness

data WorkerHeartbeat Source #

The mirror worker's consume-loop heartbeat: the wall-clock time of the worker's last recorded progress -- a successful poll of the queue, or a completed job.

It is the worker's own liveness signal, kept apart from the server's HTTP readiness so single-process health reflects a stalled worker today and a future standalone worker binary keeps the same probe. The worker advances it (via recordWorkerProgress) after each successful receive (whether or not the batch was empty -- an empty long-poll is a healthy idle, not a stall) and after each completed job, so a long batch of large artifacts cannot starve it; a liveness probe reads lastPoll and compares it against the wall clock to decide whether the loop has gone quiet for too long.

newWorkerHeartbeat :: IO WorkerHeartbeat Source #

Build a fresh WorkerHeartbeat with no poll yet recorded (lastPoll is Nothing until the worker's first successful receive).

recordPoll :: WorkerHeartbeat -> UTCTime -> IO () Source #

Stamp the heartbeat with the given instant, recording a unit of worker progress. The worker advances it (via recordWorkerProgress) after each successful receive -- the loop is alive even on an empty batch -- and after each completed job, so a long batch of large artifacts cannot starve the signal.

lastPoll :: WorkerHeartbeat -> IO (Maybe UTCTime) Source #

The instant of the worker's last recorded progress (a successful poll or a completed job), or Nothing before its first. A liveness probe reads this and compares it against the wall clock.

workerHeartbeatStaleAfter :: NominalDiffTime Source #

How long the worker's last recorded progress may be stale before the loop is considered stalled -- the staleness threshold the liveness probe applies.

The worker records progress on two events (see recordWorkerProgress): each successful poll and each completed job. The threshold must clear the larger of the two gaps. The idle gap is small -- a healthy idle worker completes a poll at least every SQS long-poll window (sqsWaitSeconds, ≤ 20s by default). The busy gap is the binding one: a single job can legitimately run a fetch and then a publish of the largest artifact the memory plan's fetch cap admits (the mirror-artifact tenant, at most 512 MiB at its ceiling), and each transfer is budgeted at the publish-visibility floor (workerPublishVisibilityBudget, ~300s for 512 MiB over a conservative ~2 MiB/s link). One healthy job therefore runs for up to about two such budgets before its heartbeat next advances.

Set above that two-budget sum (with headroom for the bounded probe, metadata re-fetch, and integrity hashing between the legs) so a healthy worker mid-large-publish is never mistaken for a stalled one. Advancing the heartbeat only once per batch under a 120s bound previously flagged such a worker dead, so an orchestrator liveness probe killed the pod mid-publish and the un-acked jobs redelivered into the identical stall: a self-inflicted restart loop. Ecluse.Worker.LivenessSpec pins the relationship to workerPublishVisibilityBudget so the two budgets cannot drift.

heartbeatHealthy :: UTCTime -> Maybe UTCTime -> Bool Source #

Whether the worker's consume loop is healthy as of now, given its last successful poll. This is the liveness signal the single-process /livez probe folds in (see Ecluse.Server), distinct from HTTP readiness.

  • Nothing (no poll yet) is healthy: the worker is still starting, not stalled.
  • A poll within workerHeartbeatStaleAfter is healthy.
  • A poll older than that is unhealthy: the loop has gone quiet for too long.
>>> import Data.Time (UTCTime (UTCTime), fromGregorian, secondsToDiffTime)
>>> let t0 = UTCTime (fromGregorian 2020 1 1) (secondsToDiffTime 0)
>>> heartbeatHealthy t0 Nothing
True
>>> let now = UTCTime (fromGregorian 2020 1 1) (secondsToDiffTime 10)
>>> heartbeatHealthy now (Just t0)
True
>>> let later = UTCTime (fromGregorian 2020 1 1) (secondsToDiffTime 700)
>>> heartbeatHealthy later (Just t0)
False

heartbeatHealthyNow :: WorkerHeartbeat -> IO Bool Source #

Read the worker heartbeat and decide liveness against the current wall clock -- the IO wrapper the liveness probe calls. True while the consume loop is alive (or still starting); False once the last successful poll is staler than workerHeartbeatStaleAfter.

Integrity verification

data IntegrityResult Source #

The result of verifying fetched bytes against the admitted integrity digests. A sum type, not a Bool, so the mismatch carries the detail an operator needs to explain why a publish was refused.

Constructors

IntegrityVerified

The bytes matched the most authoritative admitted digest.

IntegrityMismatch Text

The bytes failed the integrity gate. Carries a human-readable detail (the digest they were checked against, or that the strongest one was uncomputable).

verifyIntegrity :: NonEmpty Hash -> ByteString -> IntegrityResult Source #

Verify fetched artifact bytes against the most authoritative integrity digest the version carries -- never against a weaker one while a stronger is present.

A real npm version carries both a modern SRI sha512 digest and the legacy SHA-1 shasum. Passing on any match would let an artifact that matches the weak SHA-1 but fails the strong sha512 through -- and SHA-1 collision resistance is broken, so that is exploitable. So the gate verifies the bytes against the one digest the shared selection names (authoritativeDigest -- the same authority order the serve-side admission floor ranks by): the bytes pass iff that digest matches. A weaker digest can neither override nor rescue a failed strong one, and because the selection is shared, this gate and the admission floor can never rank the same digest set two different ways.

The bytes are recomputed in that digest's own algorithm through the shared computeDigest, the one definition of which algorithms Écluse can verify. That computable set covers every algorithm the public integrity floor admits, so an admitted artifact is always verifiable here. Each SRI Hash carries exactly one <alg>-<base64> component (mkSriHashes splits a joined wire string at construction), so the digest body compared is always a single component's, never a joined string. If the selected digest is in an algorithm the worker cannot recompute, the gate fails closed: a tampered artifact must never be admitted on the strength of a hash an attacker could forge.

This is the tamper gate before a publish: a mismatch fails the job and never publishes a corrupt or substituted artifact into the private upstream.

>>> import Ecluse.Core.Package (mkHash, HashAlg (SHA1))
>>> fmap (\h -> verifyIntegrity (h :| []) "Hello World") (mkHash SHA1 "0a4d55a8d778e5022fab701977c5d840bbc486d0")
Right IntegrityVerified
>>> fmap (\h -> verifyIntegrity (h :| []) "Hello World") (mkHash SHA1 "da39a3ee5e6b4b0d3255bfef95601890afd80709")
Right (IntegrityMismatch "the SHA1 digest did not match the fetched bytes")