acyclic_fs/kernel/
publication.rs

1//! Atomic generation publication after bounded whole-closure authentication.
2
3use super::{
4    CanonicalDecodeError, ClosureError, ClosureLimits, GenerationProof, codec::Decoder,
5    prove_generation_closure, prove_generation_closure_async, volume_authority_id,
6};
7use crate::cancellation::CancellationToken;
8use crate::foundation::{AuthorityId, Digest, Epoch, Head, OperationId, ProposedCommit, VolumeId};
9use crate::performance::{OperationFailure, WorkBudget, WorkCounters, WorkError};
10use crate::storage::{AppendOutcome, AuthorityStoreError, ObjectId, ObjectKind, PublicationPermit};
11use bytes::Bytes;
12use thiserror::Error;
13
14const PAYLOAD_DOMAIN: &[u8] = b"acyclic-fs-publish-generation-v1\0";
15const FINGERPRINT_DOMAIN: &[u8] = b"acyclic-fs-publish-fingerprint-v1\0";
16const CONTEXTUAL_FINGERPRINT_DOMAIN: &[u8] = b"acyclic-fs-publish-fingerprint-v2\0";
17
18/// Complete caller preconditions for publishing one immutable generation.
19#[derive(Clone, Copy, Debug, Eq, PartialEq)]
20pub struct PublishGenerationRequest {
21    /// Storage authority that owns this volume's ordered head.
22    pub authority_id: AuthorityId,
23    /// Volume expected inside the authenticated generation root.
24    pub volume_id: VolumeId,
25    /// Active writer-fencing epoch.
26    pub epoch: Epoch,
27    /// Exact authority head observed by the caller.
28    pub expected: Head,
29    /// Stable retry identity.
30    pub operation_id: OperationId,
31    /// Immutable generation-root object to authenticate and publish.
32    pub generation_root: ObjectId,
33}
34
35#[derive(Clone, Copy, Debug, Eq, PartialEq)]
36pub(crate) struct PublicationIntent {
37    pub(crate) operation_context: Option<Digest>,
38    pub(crate) permit: PublicationPermit,
39}
40
41impl PublicationIntent {
42    pub(crate) const fn unrestricted(operation_context: Option<Digest>) -> Self {
43        Self {
44            operation_context,
45            permit: PublicationPermit::Unrestricted,
46        }
47    }
48
49    pub(crate) const fn guarded(
50        operation_context: Option<Digest>,
51        permit: PublicationPermit,
52    ) -> Self {
53        Self {
54            operation_context,
55            permit,
56        }
57    }
58}
59
60/// Canonical semantic fact stored in one generation-publication authority record.
61#[derive(Clone, Copy, Debug, Eq, PartialEq)]
62pub struct PublishedGeneration {
63    /// Volume whose ordered authority admitted the generation.
64    pub volume_id: VolumeId,
65    /// Authenticated immutable generation-root object.
66    pub generation_root: ObjectId,
67}
68
69/// Decodes one canonical generation-publication authority payload.
70///
71/// The authority store deliberately treats payloads as opaque. This decoder is
72/// the filesystem projection boundary used to reconstruct the current volume
73/// generation without scanning immutable storage or confusing the authority
74/// hash-chain digest with a generation identity.
75///
76/// # Errors
77///
78/// Rejects the wrong domain/version, truncation, trailing bytes, an unknown
79/// object-kind tag, or any object kind other than `GenerationRoot`.
80pub fn decode_published_generation(
81    bytes: &[u8],
82    maximum_payload_bytes: u64,
83) -> Result<PublishedGeneration, CanonicalDecodeError> {
84    let mut decoder = Decoder::new(bytes, PAYLOAD_DOMAIN, 1, maximum_payload_bytes)?;
85    let volume_id = VolumeId::from_bytes(decoder.fixed()?);
86    let kind = ObjectKind::from_canonical_tag(decoder.u8()?).map_err(|_| {
87        CanonicalDecodeError::Invariant("unknown publication object kind".to_owned())
88    })?;
89    if kind != ObjectKind::GenerationRoot {
90        return Err(CanonicalDecodeError::Invariant(
91            "publication object is not a generation root".to_owned(),
92        ));
93    }
94    let generation_root = ObjectId {
95        kind,
96        digest: Digest::from_bytes(decoder.fixed()?),
97    };
98    decoder.finish()?;
99    Ok(PublishedGeneration {
100        volume_id,
101        generation_root,
102    })
103}
104
105/// Successful or semantically rejected generation publication.
106#[derive(Clone, Debug, Eq, PartialEq)]
107pub struct PublicationReceipt {
108    /// Whole-generation authentication evidence.
109    pub proof: GenerationProof,
110    /// Atomic authority result, including retry/conflict/fencing outcomes.
111    pub outcome: AppendOutcome,
112    /// Exact closure, encoding, hashing, and authority work.
113    pub work: WorkCounters,
114}
115
116/// Publication failure retaining every unit of work spent before rejection.
117pub type PublicationFailure = OperationFailure<PublicationError>;
118
119struct PreparedPublication {
120    proof: GenerationProof,
121    commit: ProposedCommit,
122    work: WorkCounters,
123}
124
125/// Fail-closed generation-publication errors.
126#[derive(Debug, Error)]
127pub enum PublicationError {
128    /// The request targeted an authority other than the volume's canonical authority.
129    #[error("authority {actual:?} does not own volume {volume_id:?}; expected {expected:?}")]
130    AuthorityMismatch {
131        /// Requested volume.
132        volume_id: VolumeId,
133        /// Sole authority derived from the volume identity.
134        expected: AuthorityId,
135        /// Authority supplied by the caller.
136        actual: AuthorityId,
137    },
138    /// The authenticated generation belongs to another volume.
139    #[error("generation belongs to volume {actual:?}, expected {expected:?}")]
140    VolumeMismatch {
141        /// Requested volume.
142        expected: VolumeId,
143        /// Volume authenticated by the generation root.
144        actual: VolumeId,
145    },
146    /// Whole-generation authentication failed.
147    #[error(transparent)]
148    Closure(#[from] ClosureError),
149    /// Staged immutable bodies could not be durably admitted before publication.
150    #[error(transparent)]
151    Objects(#[from] crate::storage::ObjectStoreError),
152    /// Atomic authority evaluation failed.
153    #[error(transparent)]
154    Authority(#[from] AuthorityStoreError),
155    /// Exact work overflowed or exceeded the admitted request budget.
156    #[error(transparent)]
157    Work(#[from] WorkError),
158}
159
160/// Authenticates an immutable generation's complete closure, then atomically
161/// submits it to the volume authority.
162///
163/// No authority operation occurs unless closure and volume validation succeed.
164/// Semantic stale-head, fenced-writer, and idempotency outcomes are returned as
165/// values so callers can resolve retries without guessing whether publication
166/// became durable.
167///
168/// # Errors
169///
170/// Fails with an exact work receipt on closure, volume, authority-backend, or
171/// work-admission failure.
172pub fn publish_generation<O: crate::ImmediateObjectStore, A: crate::ImmediateAuthorityStore>(
173    objects: &O,
174    authority: &A,
175    request: PublishGenerationRequest,
176    closure_limits: ClosureLimits,
177    budget: WorkBudget,
178) -> Result<PublicationReceipt, PublicationFailure> {
179    validate_authority(request)?;
180    let proof = prove_generation_closure(objects, request.generation_root, closure_limits, budget)
181        .map_err(|failure| OperationFailure::new(failure.error.into(), *failure.work))?;
182    let prepared = prepare_publication(proof, request, None, budget)?;
183    let mut work = prepared.work;
184    let remaining = work
185        .remaining(budget)
186        .map_err(|error| OperationFailure::new(error.into(), work))?;
187    let receipt = authority
188        .compare_and_append(
189            request.authority_id,
190            request.epoch,
191            request.expected,
192            prepared.commit,
193            remaining,
194        )
195        .map_err(|failure| failure.map_with_prior_work(work, Into::into))?;
196    work = work
197        .checked_add(receipt.work)
198        .map_err(|error| OperationFailure::new(error.into(), work))?;
199    work.verify(budget)
200        .map_err(|error| OperationFailure::new(error.into(), work))?;
201    Ok(PublicationReceipt {
202        proof: prepared.proof,
203        outcome: receipt.value,
204        work,
205    })
206}
207
208/// Asynchronously authenticates and atomically publishes one generation.
209///
210/// This is the browser/remote execution surface for the same canonical closure
211/// proof, publication payload, fingerprint, and authority state machine used by
212/// [`publish_generation`].
213///
214/// # Errors
215///
216/// Returns exact closure, cancellation, authority, and work failures without
217/// exposing a generation before its complete immutable closure is proven.
218pub async fn publish_generation_async<O: crate::AsyncObjectStore, A: crate::AsyncAuthorityStore>(
219    objects: &O,
220    authority: &A,
221    request: PublishGenerationRequest,
222    closure_limits: ClosureLimits,
223    budget: WorkBudget,
224    cancellation: &CancellationToken,
225) -> Result<PublicationReceipt, PublicationFailure> {
226    publish_generation_async_inner(
227        objects,
228        authority,
229        request,
230        PublicationIntent::unrestricted(None),
231        closure_limits,
232        budget,
233        cancellation,
234    )
235    .await
236}
237
238pub(crate) async fn publish_generation_async_with_context<
239    O: crate::AsyncObjectStore,
240    A: crate::AsyncAuthorityStore,
241>(
242    objects: &O,
243    authority: &A,
244    request: PublishGenerationRequest,
245    operation_context: Digest,
246    closure_limits: ClosureLimits,
247    budget: WorkBudget,
248    cancellation: &CancellationToken,
249) -> Result<PublicationReceipt, PublicationFailure> {
250    publish_generation_async_inner(
251        objects,
252        authority,
253        request,
254        PublicationIntent::unrestricted(Some(operation_context)),
255        closure_limits,
256        budget,
257        cancellation,
258    )
259    .await
260}
261
262pub(crate) async fn publish_generation_async_with_permit<
263    O: crate::AsyncObjectStore,
264    A: crate::AsyncAuthorityStore,
265>(
266    objects: &O,
267    authority: &A,
268    request: PublishGenerationRequest,
269    intent: PublicationIntent,
270    closure_limits: ClosureLimits,
271    budget: WorkBudget,
272    cancellation: &CancellationToken,
273) -> Result<PublicationReceipt, PublicationFailure> {
274    publish_generation_async_inner(
275        objects,
276        authority,
277        request,
278        intent,
279        closure_limits,
280        budget,
281        cancellation,
282    )
283    .await
284}
285
286async fn publish_generation_async_inner<
287    O: crate::AsyncObjectStore,
288    A: crate::AsyncAuthorityStore,
289>(
290    objects: &O,
291    authority: &A,
292    request: PublishGenerationRequest,
293    intent: PublicationIntent,
294    closure_limits: ClosureLimits,
295    budget: WorkBudget,
296    cancellation: &CancellationToken,
297) -> Result<PublicationReceipt, PublicationFailure> {
298    validate_authority(request)?;
299    let proof = prove_generation_closure_async(
300        objects,
301        request.generation_root,
302        closure_limits,
303        budget,
304        cancellation,
305    )
306    .await
307    .map_err(|failure| OperationFailure::new(failure.error.into(), *failure.work))?;
308    let operation_context = permit_context(intent.operation_context, intent.permit);
309    let prepared = prepare_publication(proof, request, operation_context, budget)?;
310    let mut work = prepared.work;
311    // The authority CAS is the only point at which immutable bodies become
312    // reachable after a crash. Drain any private staged bodies first; a failed
313    // drain leaves no published generation and can be retried idempotently.
314    let drained = objects
315        .flush_before_publish(
316            work.remaining(budget)
317                .map_err(|error| OperationFailure::new(error.into(), work))?,
318            cancellation,
319        )
320        .await
321        .map_err(|failure| failure.map_with_prior_work(work, Into::into))?;
322    work = work
323        .checked_add(drained.work)
324        .map_err(|error| OperationFailure::new(error.into(), work))?;
325    let remaining = work
326        .remaining(budget)
327        .map_err(|error| OperationFailure::new(error.into(), work))?;
328    let receipt = crate::AsyncAuthorityStore::compare_and_append_guarded(
329        authority,
330        crate::GuardedAppend {
331            authority_id: request.authority_id,
332            epoch: request.epoch,
333            expected: request.expected,
334            commit: prepared.commit,
335            permit: intent.permit,
336        },
337        remaining,
338        cancellation,
339    )
340    .await
341    .map_err(|failure| failure.map_with_prior_work(work, Into::into))?;
342    work = work
343        .checked_add(receipt.work)
344        .map_err(|error| OperationFailure::new(error.into(), work))?;
345    work.verify(budget)
346        .map_err(|error| OperationFailure::new(error.into(), work))?;
347    Ok(PublicationReceipt {
348        proof: prepared.proof,
349        outcome: receipt.value,
350        work,
351    })
352}
353
354fn permit_context(context: Option<Digest>, permit: PublicationPermit) -> Option<Digest> {
355    match permit {
356        PublicationPermit::Unrestricted => context,
357        PublicationPermit::Lease {
358            authority_id,
359            workspace_id,
360            lease_id,
361            expires_at_millis,
362        } => {
363            let mut hasher = blake3::Hasher::new();
364            hasher.update(b"acyclic-fs-publication-lease-v1\0");
365            match context {
366                Some(context) => {
367                    hasher.update(&[1]);
368                    hasher.update(context.as_bytes());
369                }
370                None => {
371                    hasher.update(&[0]);
372                }
373            };
374            hasher.update(&authority_id);
375            hasher.update(&workspace_id);
376            hasher.update(&lease_id);
377            hasher.update(&expires_at_millis.to_be_bytes());
378            Some(Digest::from_bytes(*hasher.finalize().as_bytes()))
379        }
380        PublicationPermit::Reservation {
381            operation_id,
382            gate_tail,
383            expected,
384        } => {
385            let mut hasher = blake3::Hasher::new();
386            hasher.update(b"acyclic-fs-publication-reservation-v1\0");
387            match context {
388                Some(context) => {
389                    hasher.update(&[1]);
390                    hasher.update(context.as_bytes());
391                }
392                None => {
393                    hasher.update(&[0]);
394                }
395            };
396            hasher.update(&operation_id);
397            hasher.update(&gate_tail.to_be_bytes());
398            hasher.update(&expected.epoch.get().to_be_bytes());
399            hasher.update(&expected.sequence.get().to_be_bytes());
400            hasher.update(expected.digest.as_bytes());
401            Some(Digest::from_bytes(*hasher.finalize().as_bytes()))
402        }
403    }
404}
405
406fn prepare_publication(
407    proof: GenerationProof,
408    request: PublishGenerationRequest,
409    operation_context: Option<Digest>,
410    budget: WorkBudget,
411) -> Result<PreparedPublication, PublicationFailure> {
412    let mut work = proof.work;
413    if proof.root.volume_id != request.volume_id {
414        return Err(OperationFailure::new(
415            PublicationError::VolumeMismatch {
416                expected: request.volume_id,
417                actual: proof.root.volume_id,
418            },
419            work,
420        ));
421    }
422    let payload_bytes = u64::try_from(publication_payload_length()).unwrap_or(u64::MAX);
423    let fingerprint_bytes =
424        u64::try_from(fingerprint_input_length(operation_context)).unwrap_or(u64::MAX);
425    let fingerprint_domain = fingerprint_domain(operation_context);
426    let peak_allocation_bytes = payload_bytes
427        .checked_add(fingerprint_bytes)
428        .ok_or_else(|| OperationFailure::new(WorkError::Overflow.into(), work))?;
429    let semantic_work = WorkCounters {
430        bytes_encoded: payload_bytes
431            .checked_add(fingerprint_bytes)
432            .ok_or_else(|| OperationFailure::new(WorkError::Overflow.into(), work))?,
433        bytes_hashed: u64::try_from(fingerprint_domain.len())
434            .unwrap_or(u64::MAX)
435            .checked_add(fingerprint_bytes)
436            .ok_or_else(|| OperationFailure::new(WorkError::Overflow.into(), work))?,
437        bytes_copied: payload_bytes,
438        allocation_operations: 2,
439        peak_allocation_bytes,
440        ..WorkCounters::default()
441    };
442    work = work
443        .checked_add(semantic_work)
444        .map_err(|error| OperationFailure::new(error.into(), work))?;
445    work.verify(budget)
446        .map_err(|error| OperationFailure::new(error.into(), work))?;
447    let payload = encode_publication_payload(request.volume_id, request.generation_root);
448    let fingerprint = publication_fingerprint(request, operation_context, &payload);
449    Ok(PreparedPublication {
450        proof,
451        commit: ProposedCommit {
452            operation_id: request.operation_id,
453            fingerprint,
454            payload: Bytes::from(payload),
455        },
456        work,
457    })
458}
459
460#[cfg(all(feature = "native-watch", not(target_arch = "wasm32")))]
461pub(crate) fn contextual_publication_fingerprint(
462    request: PublishGenerationRequest,
463    operation_context: Digest,
464) -> Digest {
465    let payload = encode_publication_payload(request.volume_id, request.generation_root);
466    publication_fingerprint(request, Some(operation_context), &payload)
467}
468
469fn publication_fingerprint(
470    request: PublishGenerationRequest,
471    operation_context: Option<Digest>,
472    payload: &[u8],
473) -> Digest {
474    let fingerprint_input = encode_fingerprint_input(&request, operation_context, payload);
475    let mut hasher = blake3::Hasher::new();
476    hasher.update(fingerprint_domain(operation_context));
477    hasher.update(&fingerprint_input);
478    Digest::from_bytes(*hasher.finalize().as_bytes())
479}
480
481fn validate_authority(request: PublishGenerationRequest) -> Result<(), PublicationFailure> {
482    let expected = volume_authority_id(request.volume_id);
483    if request.authority_id == expected {
484        Ok(())
485    } else {
486        Err(OperationFailure::before_work(
487            PublicationError::AuthorityMismatch {
488                volume_id: request.volume_id,
489                expected,
490                actual: request.authority_id,
491            },
492        ))
493    }
494}
495
496const fn publication_payload_length() -> usize {
497    PAYLOAD_DOMAIN.len() + 2 + 16 + 1 + 32
498}
499
500const fn fingerprint_input_length(operation_context: Option<Digest>) -> usize {
501    16 + 8
502        + 8
503        + 8
504        + 32
505        + publication_payload_length()
506        + if operation_context.is_some() { 32 } else { 0 }
507}
508
509pub(crate) fn encode_publication_payload(
510    volume_id: VolumeId,
511    generation_root: ObjectId,
512) -> Vec<u8> {
513    let mut bytes = Vec::with_capacity(publication_payload_length());
514    bytes.extend_from_slice(PAYLOAD_DOMAIN);
515    bytes.extend_from_slice(&1_u16.to_le_bytes());
516    bytes.extend_from_slice(&volume_id.into_bytes());
517    bytes.push(generation_root.kind.canonical_tag());
518    bytes.extend_from_slice(generation_root.digest.as_bytes());
519    bytes
520}
521
522fn fingerprint_domain(operation_context: Option<Digest>) -> &'static [u8] {
523    if operation_context.is_some() {
524        CONTEXTUAL_FINGERPRINT_DOMAIN
525    } else {
526        FINGERPRINT_DOMAIN
527    }
528}
529
530fn encode_fingerprint_input(
531    request: &PublishGenerationRequest,
532    operation_context: Option<Digest>,
533    payload: &[u8],
534) -> Vec<u8> {
535    let mut bytes = Vec::with_capacity(fingerprint_input_length(operation_context));
536    bytes.extend_from_slice(&request.authority_id.into_bytes());
537    bytes.extend_from_slice(&request.epoch.get().to_le_bytes());
538    bytes.extend_from_slice(&request.expected.epoch.get().to_le_bytes());
539    bytes.extend_from_slice(&request.expected.sequence.get().to_le_bytes());
540    bytes.extend_from_slice(request.expected.digest.as_bytes());
541    if let Some(context) = operation_context {
542        bytes.extend_from_slice(context.as_bytes());
543    }
544    bytes.extend_from_slice(payload);
545    bytes
546}
547
548#[cfg(all(test, feature = "memory"))]
549#[path = "tests/publication.rs"]
550mod tests;
acyclic_fs/kernel/publication.rs

acyclic_fs/kernel/
publication.rs
