Struct Pipeline 

pub struct Pipeline<B, N, H = Hooks> { /* private fields */ }

The request pipeline over blobs B, metadata N and hooks H.

Implementations

impl<B: MultipartBlobStore, N: NamespaceStore, H: HookSet> Pipeline<B, N, H>

pub async fn get_authority_generation( &self, namespace: &str, ) -> Result<u64, ServerError>

Read the optional namespace authority generation outside auth-v2.

Errors

Disabled fencing, invalid namespace or storage failure.

pub async fn set_authority_generation( &self, wire: &str, ) -> Result<u64, ServerError>

Verify a deployment statement, install its target and finish the lease barrier. Completion retries are idempotent; each call scans at most one four-shard slice.

Errors

Disabled configuration, rejected statement or pending completion (Retry-After: 1).

impl<B: MultipartBlobStore, N: NamespaceStore, H: HookSet> Pipeline<B, N, H>

pub async fn begin_upload( &self, a: &Authenticated, ref_name: &str, pack_id: &[u8], bytes: u64, ) -> Result<BeginUploadResult, ServerError>

Open a stateless authenticated upload ticket in the target ref shard.

Errors

Invalid geometry/ref, unsupported auth or missing keys, policy or cap refusal, and the usual replay, quota, lease and storage errors.

pub async fn begin_upload_with_meta( &self, a: &Authenticated, ref_name: &str, pack_id: &[u8], bytes: u64, ) -> Result<(BeginUploadResult, ResponseMeta), ServerError>

Begin a resumable upload and return success-only admission headers.

impl<B: MultipartBlobStore, N: NamespaceStore, H: HookSet> Pipeline<B, N, H>

pub async fn get_grant_epoch(&self, namespace: &str) -> Result<u64, ServerError>

Read a namespace epoch without authentication or repository resolution.

Errors

invalid_argument for a bad namespace, unimplemented under Single, or a mapped storage error.

pub async fn set_grant_epoch( &self, signed_statement: &str, ) -> Result<u64, ServerError>

Verify an owner epoch statement, CAS the coordinator epoch, then fence every leased shard before reporting completion.

Errors

permission_denied for a rejected statement, unimplemented under Single, or unavailable while completion is pending.

impl<B: MultipartBlobStore, N: NamespaceStore + Clone + 'static, H: HookSet> Pipeline<B, N, H>

pub fn http_objects_enabled(&self) -> bool

Whether this pipeline can mount HTTP object routes. Opaque pipelines cannot.

pub fn url_token_config(&self) -> Option<&UrlTokenConfig>

Public-key publication configuration; never exposes seeds.

pub fn with_http_seams(self, edit: impl FnOnce(HttpSeams) -> HttpSeams) -> Self

Replace the inert seams of an HTTP-objects pipeline: admission (WP-4.13), proofs (WP-4.14b), tokens (WP-4.15), takedown (WP-5.9a) or a maintained reachable set (WP-5.3a). A pipeline built without PipelineConfig::http_objects is unchanged.

pub async fn serve_http_object( &self, req: &HttpObjectRequest<'_>, ) -> HttpObjectResponse

Serve one HTTP object request (GET, HEAD or OPTIONS) and never fail: every error is mapped to its §3 response. The principal is always anonymous, whatever credentials the request carries (§7); the raw path, query and any token are never logged.

pub async fn serve_http_object_with_runtime( &self, req: &HttpObjectRequest<'_>, runtime: HttpReadRuntime, ) -> HttpObjectResponse

Adapter-provided retained settlement runtime for this request.

pub async fn serve_http_object_with_proofs( &self, req: &HttpObjectRequest<'_>, runtime: HttpReadRuntime, proofs: Arc<dyn ProofServer>, ) -> HttpObjectResponse

Adapter-provided proof builder and retained settlement runtime.

impl<B: MultipartBlobStore, N: NamespaceStore + Clone + 'static, H: HookSet> Pipeline<B, N, H>

pub async fn object_reader<'a>( &'a self, repo: RepoId, view: ReaderView<'a>, ) -> Result<ObjectReader<'a, B, N, H>, ServerError>

Construct a repository reader using verified owner authority.

Errors

Missing indexed/HTTP configuration, or invalid/non-owner credentials.

impl<B: MultipartBlobStore, N: NamespaceStore, H: HookSet> Pipeline<B, N, H>

pub fn server_info(&self) -> ServerInfo

Configured deployment information. Never resolves a repository or reads a store; the atomic flag comes from store capabilities only.

impl<B: MultipartBlobStore, N: NamespaceStore, H: HookSet> Pipeline<B, N, H>

pub async fn list_repos_page( &self, a: &Authenticated, namespace: &str, name_prefix: &str, page_size: Option<u32>, token: Option<&[u8]>, ) -> Result<RepoPage, ServerError>

List a namespace without visiting private rows for the public view. Signed envelopes use the ordinary X-Repository syntax; its namespace must match namespace. The name is only a namespace selector for this RPC. Grants never extend listing rights. Check and Authority hooks concern the entire namespace.

Errors

Invalid namespace/prefix/size/token, mismatched authentication, or unavailable storage/hooks.

impl<B: MultipartBlobStore, N: NamespaceStore, H: HookSet> Pipeline<B, N, H>

pub async fn open_part( &self, a: &Authenticated, token: &[u8], index: u32, ) -> Result<PartUploadSession<'_, B, N, H>, ServerError>

Validate a part header and ticket before opening any part sink.

Errors

Invalid or mismatched ticket, commitment or part geometry; storage errors while opening the part.

pub async fn complete_upload( &self, a: &Authenticated, token: &[u8], receipts: &[Vec<u8>], ) -> Result<(), ServerError>

Authenticate all receipts and the merged pack root before any store call; completion writes no metadata rows.

Errors

Invalid ticket, receipt, root, length or storage session.

impl<B: MultipartBlobStore, N: NamespaceStore, H: HookSet> Pipeline<B, N, H>

pub async fn plan_repository_purge( &self, partition: &Partition, repo: &RepoId, trigger: Trigger, operation_id: &str, now_ms: u64, ) -> Result<Batch, ServerError>

Plan an audited safety purge in the same batch as a serving stop. operation_id is stable across retries of the automatic action. Callers merge this batch into the authoritative state mutation.

pub async fn invalidate_local_cache(&self, repo: &RepoId)

Immediately invalidate local entries after committing a serving stop and its durable purge work. Timer 11 retries any failed cache deletion.

impl<B: MultipartBlobStore, N: NamespaceStore, H: HookSet> Pipeline<B, N, H>

pub async fn bump_epoch( &self, ns: &NamespaceKey, new_epoch: u64, ) -> Result<(), ServerError>

Advance the coordinator epoch, serialized with every epoch lease grant.

Errors

invalid_argument unless the increment is in 1..=1024; unavailable on repeated contention, or a mapped storage failure.

pub async fn mark_lease_table_recovered( &self, ns: &NamespaceKey, ) -> Result<(), ServerError>

Declare lease-table recovery using the real pipeline clock. Any procedure that restores or rebuilds a coordinator partition MUST call this before serving writes for that namespace. The durable marker prevents completion for epoch_lease + margin after recovery.

Errors

A mapped storage failure, or internal for an impossible batch outcome.

pub async fn revoke_step( &self, ns: &NamespaceKey, budget: &RevokeBudget, ) -> Result<RevokeProgress, ServerError>

Push and acknowledge at most four live shards in one bounded slice. Outside a declared recovery hold-off, ls.acked_epoch = n only if the shard’s el durably holds epoch >= n, or all older-epoch writes are already past their commit deadline. Recovery fences surviving copies until then. Renewal alone never raises a live row’s acknowledgement.

Errors

A mapped storage or codec failure. Contention leaves progress pending.

impl<B: MultipartBlobStore, N: NamespaceStore, H: HookSet> Pipeline<B, N, H>

pub fn scanner_retrieval_enabled(&self) -> bool

Whether adapters may mount the private route. Default false.

pub async fn retrieve_scanner_pack( &self, body: &[u8], headers: &Headers, ) -> Result<RetrievalResponse, ServerError>

Verify both credentials, strongly check existing bound ticket state and global denial, then read one bounded raw pack range. No public membership authorization or pack decoding is involved.

Errors

Uniform not_found, including storage errors and exhausted budgets.

impl<B: BlobStore, N: NamespaceStore, H: HookSet> Pipeline<B, N, H>

pub async fn namespace_relay_watermark( &self, ns: &NamespaceKey, ) -> Result<u64, ServerError>

Namespace lower bound on commit time of undelivered relay rows for native GC and takedown. Consumers compare it against T + MAX_APPLY_WINDOW + margin. Workers use the resumable store step. A new shard’s stale-low first report or recovery can lower the value; it is unavailable until the recovered lease table is reconciled.

Errors

Unavailable during lease-table recovery or on an unreadable store.

pub async fn active_shards( &self, ns: &NamespaceKey, cursor: Option<&Cursor>, limit: u32, ) -> Result<ActiveShardsPage, ServerError>

Enumerate coordinator ref-shard rows for a safety re-check. Expired rows retained for relay backlog are included.

Errors

Unavailable during lease-table recovery or on an unreadable store.

impl<B: MultipartBlobStore, N: NamespaceStore, H: HookSet> Pipeline<B, N, H>

pub fn new( blobs: B, meta: N, hooks: H, cfg: PipelineConfig, clock: Arc<dyn Clock>, metrics: Arc<dyn Metrics>, ) -> Result<Self, ServerError>

A pipeline over blobs and meta, routed by cfg.sharding.

Errors

invalid_argument for a configuration the store cannot serve: auth v2 needs every key class and atomic multi-key batches (so FsLayoutStore never runs auth v2); a store without atomic batches must report an implicit layout version; a store’s layout version must be this binary’s; the page limit and apply window must be positive. Resumable upload and advertised page limits must be valid and the part capacity must reach the upload byte cap. Namespace/write policy combinations must be compatible; any requires non-default admission or its explicit unsafe override, and an authority authorizer must not be the open default.

pub fn inspection_max_objects(&self) -> Option<u32>

Total launch inspected-set cap, absent when inspection is disabled.

pub fn with_inspectors( self, inspectors: Vec<Arc<dyn ContentInspector>>, batch_max: usize, ) -> Result<Self, ServerError>

Configure the launch’s synchronous, fail-closed inspectors.

Errors

Invalid launch settings or a deployment without indexed, ticketed, restricted writes and atomic metadata.

pub fn with_write_gate(self) -> Self

Run the writes to one partition one at a time in this process: each write’s read-plan-apply loop waits for the partition’s gate, as a Durable Object’s input gate serializes them on Workers. For a single-process server over a single-writer store (native SQLite, which commits one batch at a time anyway): concurrent writes that share a key, such as one signer’s quota window, then never exhaust the re-plan bound and fail aborted. D34 lease-grant batches also take this gate after admission. Reads are not gated. Several processes on one store still race, through the optimistic loop.

pub fn with_auth(&self, auth: AuthMode) -> Result<Self, ServerError>

where B: Clone, N: Clone, H: Clone,

A second pipeline over the same stores, hooks, shard map, clock, metrics, test fault hooks and write gate, authenticating with auth: how one server hosts bindings with different identity sources on one root (an enc listener’s TransportIdentity beside an HTTP listener’s bearer token or auth v2) while its writes to a partition still pass one gate. Every other setting is self’s.

Errors

As Self::new for auth over these stores.

pub fn authenticate( &self, meta: &RequestMeta<'_>, ) -> Result<Authenticated, ServerError>

Stages 0a and 1: verify credentials and map the identity. Pure and synchronous; writes no state. The result is bound to meta.procedure. Under test-faults it also reads the request’s test directives: the clock skew shifts business time, including the auth v2 validity window, for this request only.

Errors

unauthenticated for missing or invalid credentials; invalid_argument for a malformed test directive. A rejection is recorded like any failed request (procedure, code, latency), with principal none: no entry point runs after it to record it.

pub async fn list_refs( &self, a: &Authenticated, prefix: &str, ) -> Result<Vec<RefEntry>, ServerError>

Every ref of the repository under prefix at a path-component boundary, with the prefix and its / stripped (SPEC-REFS §4, see refs::list_scan_prefix), read page by page.

Errors

not_found for a nonexistent Multi repository; invalid_argument for an invalid prefix or one over refs::MAX_REF_NAME_BYTES; the authorizer’s error; unavailable for a bucket scan failure.

pub fn with_published_source(self, source: Arc<dyn PublishedSource>) -> Self

Attach an explicit published reader source (snapshot opt-in).

pub fn with_publication_policy( self, policy: Arc<dyn PublicationPolicy>, ) -> Result<Self, ServerError>

Install the inspection preparation and immediate hold gate (post-launch WP-5.5c). Inspection requires indexed mode and an owner/authority write policy.

pub async fn read_ref( &self, a: &Authenticated, name: &str, ) -> Result<Option<Hash>, ServerError>

One ref’s id, if it exists.

Errors

not_found for a nonexistent Multi repository; invalid_argument for an invalid name; the authorizer’s error; internal for a storage failure.

pub async fn update_ref( &self, a: &Authenticated, upd: RefUpdate, ) -> Result<UpdateRefResult, ServerError>

Compare-and-swap one ref. A conflict is a result, not an error.

Errors

See the stage functions; a stored rejection comes back as its error.

pub async fn update_ref_with_meta( &self, a: &Authenticated, upd: RefUpdate, ) -> Result<(UpdateRefResult, ResponseMeta), ServerError>

Compare-and-swap a ref and return success-only admission headers.

pub async fn advance_refs( &self, a: &Authenticated, head: RefUpdate, packmap: RefUpdate, ) -> Result<AdvanceOutcome, ServerError>

Advance a branch head and its packmap together: one batch on an atomic store, else packmap then head (Transport::advance_refs’s default order).

Errors

See the stage functions; a stored rejection comes back as its error.

pub async fn advance_refs_with_meta( &self, a: &Authenticated, head: RefUpdate, packmap: RefUpdate, ) -> Result<(AdvanceOutcome, ResponseMeta), ServerError>

Advance refs and return success-only admission headers.

pub async fn advance_refs_with_tickets( &self, a: &Authenticated, head: RefUpdate, packmap: RefUpdate, tickets: Vec<Hash>, ) -> Result<AdvanceOutcome, ServerError>

Advance both refs while consuming the named upload tickets.

Errors

Invalid ticket bindings, incomplete uploads, authorization or storage errors.

pub async fn advance_refs_with_tickets_with_meta( &self, a: &Authenticated, head: RefUpdate, packmap: RefUpdate, tickets: Vec<Hash>, ) -> Result<(AdvanceOutcome, ResponseMeta), ServerError>

Advance refs with tickets and return success-only admission headers.

pub async fn pack_exists( &self, a: &Authenticated, key: PackKey, ) -> Result<bool, ServerError>

Whether the pack is present in a Single deployment’s blob store. Multi deployments require repository membership before serving packs.

Errors

not_found for a nonexistent Multi repository; the authorizer’s error; internal for a storage failure.

pub async fn issue_object_url( &self, a: &Authenticated, target: UrlTarget, ttl_seconds: u32, ) -> Result<MintedToken, ServerError>

IssueObjectUrl (SPEC-WRITE-GRANTS §9.4): mint a token binding the deployment’s audience, the request’s repository identity and target at the stored grant epoch. The target is never resolved — serving decides what it names, so minting reveals nothing about the repository’s contents. The token is a credential: it is never logged.

Errors

unimplemented when no URL-token key is configured, before any repository access; unauthenticated for an unsigned request (stage 0 already rejects it under auth v2); the read errors of Self::authorize_read, including the uniform not_found on a private repository.

pub async fn set_repo_visibility( &self, a: &Authenticated, req: VisibilityRequest, ) -> Result<(), ServerError>

SetRepoVisibility (SPEC-WRITE-GRANTS §9.1): the envelope mode is a signed, replay-protected write of the rv row; the statement mode verifies an unsigned owner-signed statement and keeps the newest created. Only deployments where Self::visibility_applies holds have visibility at all.

Errors

unauthenticated for an unsigned envelope request, invalid_argument for a statement on a signed request, failed_precondition when the deployment has no repository visibility, permission_denied for a grant, a non-owner, an unserved namespace or a rejected statement.

pub async fn open_upload( &self, a: &Authenticated, pack_id: Option<&[u8]>, total_bytes: Option<u64>, ) -> Result<UploadSession<'_, B, N, H>, ServerError>

Stages 0–3 of an UploadPack whose header declared pack_id and total_bytes (None when absent): framing, the signed pack: commitment, the replay lookup, authorization, admission and, for a new signed operation, the reservation. Nothing is read from the stream before it returns.

Errors

failed_precondition when the advertised threshold requires a ticket; the header’s crate::upload::UploadError; unauthenticated when the header differs from the signed commitment; a stored or in-flight replay answer; a hook’s error; the reservation’s error.

pub async fn open_ticketed_upload( &self, a: &Authenticated, pack_id: Option<&[u8]>, total_bytes: Option<u64>, token: &[u8], ) -> Result<UploadSession<'_, B, N, H>, ServerError>

Open a stateless ticketed UploadPack stream after checking the header, signed commitment and token, before reading any body bytes.

Errors

Framing, authentication, binding and blob-store failures.

pub async fn download( &self, a: &Authenticated, key: PackKey, ) -> Result<DownloadStream, ServerError>

A pack’s bytes as chunks of at most download_chunk_max bytes.

Errors

not_found for a missing repository or non-member pack, before any chunk; the authorizer’s error; internal for a storage failure.

The request is recorded ok when the last chunk is yielded, with its error at the first failure, and as canceled when the stream is dropped before either.

pub async fn health(&self) -> HealthStatus

Probe both stores.

pub fn auth_mode(&self) -> &AuthMode

How this pipeline authenticates (the ssh session requires TransportIdentity; an HTTP adapter may pre-check a bearer token before it spends resources on the request).

pub fn capabilities(&self) -> PipelineCapabilities

What this pipeline offers.

pub fn with_header( &self, err: ServerError, name: &str, value: &str, ) -> ServerError

Add a response header to err, counting a dropped one in METRIC_HEADER_DROPPED (what ServerError::with_header cannot do without a metrics sink). The value is never logged.

Trait Implementations

impl<B, N, H> Debug for Pipeline<B, N, H>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.