flint_lean/
inbox.rs

1//! The request cell: ONE CAS document per workspace carrying the two verb
2//! requests a party outside the pod can make (plan §2.5) — "please
3//! publish" and "please pull".
4//!
5//! It was also the HITL inbox and the barrier-window token. Under P2
6//! (simplification step 5, 2026-09-25) a UI write COMMITS — the gateway
7//! CASes the manifest itself (`manifest::commit_edit`) — so there are no
8//! entries, no declared removals and no window any more.
9
10use serde::{Deserialize, Serialize};
11
12use flint_store::{
13    crc64_nvme, GenerationStamps, ObjectStore, PutCondition, StoreError,
14};
15
16use super::{now_unix, LeanConfig, LeanError, LeanResult};
17
18/// A verb asked for through the gateway door (§2.5, D14). Idempotent
19/// STATE, not a queue: repeated sets before the syncer acts collapse
20/// to the newest, which is why neither field needs a rate limit, an
21/// exactly-once protocol, or a clearing CAS of its own.
22#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
23pub struct VerbRequest {
24    pub requested_unix: u64,
25    pub requestor: String,
26}
27
28#[derive(Debug, Clone, Default, Serialize, Deserialize)]
29pub struct InboxDoc {
30    /// "Please publish" from outside the pod (§2.5).
31    #[serde(default, skip_serializing_if = "Option::is_none")]
32    pub boundary_request: Option<VerbRequest>,
33    /// "Please pull" from outside the pod — CARRIED, never performed
34    /// (D14). A boundary publishes what is already on disk and touches
35    /// no local file; `sync` re-derives the tree against the current
36    /// remote manifest and DELETES local files for remotely-deleted
37    /// paths. Performing that on a remote's say-so would upgrade what a
38    /// leaked gateway bearer can do from "publish, plus hand over these
39    /// N named objects" to "rewrite and delete across a running agent's
40    /// tree, at my timing, under a scope I choose".
41    #[serde(default, skip_serializing_if = "Option::is_none")]
42    pub sync_request: Option<VerbRequest>,
43}
44
45pub struct LoadedInbox {
46    pub doc: InboxDoc,
47    /// None ⇒ the cell does not exist yet (first CAS is If-None-Match:*).
48    pub etag: Option<String>,
49}
50
51pub async fn load(store: &dyn ObjectStore, cfg: &LeanConfig) -> LeanResult<LoadedInbox> {
52    match store.get_whole(&cfg.inbox_key(), None).await {
53        Ok((meta, bytes)) => {
54            let doc = serde_json::from_slice(&bytes)
55                .map_err(|e| LeanError::State(format!("inbox parse: {e}")))?;
56            Ok(LoadedInbox { doc, etag: Some(meta.etag) })
57        }
58        Err(StoreError::NotFound(_)) => Ok(LoadedInbox { doc: InboxDoc::default(), etag: None }),
59        Err(e) => Err(e.into()),
60    }
61}
62
63/// A CAS on the cell that lost to another writer: 412, or S3's 409
64/// ConditionalRequestConflict — what S3 answers a conditional write that
65/// races another on the same key, and this is the most contended key a
66/// workspace has. Every loop below re-reads and retries both. Matching
67/// 412 alone failed the gateway's append after a LANDED UI write, and the
68/// earlier acked write that one replaced was dropped as superseded at the
69/// next consume (review 2026-09-18, H3). The lease and the manifest CAS
70/// always treated the pair alike.
71fn lost_race(e: &LeanError) -> bool {
72    matches!(e, LeanError::Store(StoreError::PreconditionFailed(_) | StoreError::Conflict(_)))
73}
74
75pub async fn cas_write(
76    store: &dyn ObjectStore,
77    cfg: &LeanConfig,
78    doc: &InboxDoc,
79    expected: Option<&str>,
80    epoch: u64,
81) -> LeanResult<String> {
82    let bytes = serde_json::to_vec_pretty(doc)
83        .map_err(|e| LeanError::State(format!("inbox: {e}")))?;
84    let crc = crc64_nvme(&bytes);
85    let cond = match expected {
86        Some(etag) => PutCondition::IfMatch(etag.to_string()),
87        None => PutCondition::IfNoneMatchAny,
88    };
89    let stamps = GenerationStamps {
90        generation: 0,
91        epoch,
92        flush_uuid: "inbox".into(),
93        boundary_source: None,
94        posix: None,
95    };
96    let meta = store.put_whole(&cfg.inbox_key(), bytes.into(), &cond, &stamps, crc).await?;
97    Ok(meta.etag)
98}
99
100/// Which verb a gateway request is asking for.
101#[derive(Debug, Clone, Copy, PartialEq, Eq)]
102pub enum RequestedVerb {
103    Boundary,
104    Sync,
105}
106
107/// The GATEWAY side of §2.5: set one of the two request fields under
108/// the same CAS discipline every other inbox write uses.
109pub async fn gateway_request(
110    store: &dyn ObjectStore,
111    cfg: &LeanConfig,
112    verb: RequestedVerb,
113    requestor: &str,
114) -> LeanResult<VerbRequest> {
115    let req = VerbRequest {
116        requested_unix: now_unix(),
117        requestor: requestor.chars().take(128).collect(),
118    };
119    for _ in 0..5 {
120        let loaded = load(store, cfg).await?;
121        let mut doc = loaded.doc;
122        // Newest wins: the field is state, so a burst collapses instead
123        // of queueing. This is what makes a rate limit unnecessary on
124        // the transport (the HONOR is still min-interval'd and budgeted
125        // like any other sentinel).
126        match verb {
127            RequestedVerb::Boundary => doc.boundary_request = Some(req.clone()),
128            RequestedVerb::Sync => doc.sync_request = Some(req.clone()),
129        }
130        match cas_write(store, cfg, &doc, loaded.etag.as_deref(), 0).await {
131            Ok(_) => return Ok(req),
132            Err(e) if lost_race(&e) => continue,
133            Err(e) => return Err(e),
134        }
135    }
136    Err(LeanError::State("inbox verb request lost 5 CAS races".into()))
137}
flint_lean/inbox.rs

flint_lean/
inbox.rs
