flint-lean 0.11.0

//! The subtree lease, held for ONE BARRIER's commit section (design
//! 2026-09-13, `docs/plans/flint-lean-writer-lease-and-gated-assessment.md`
//! §4), with a FIFO ticket so W writers wait at most W barriers.
//!
//! What the cell arbitrates is who may INSTALL a manifest. Nothing
//! else: checkout, sync, status and the uploads of a barrier hold
//! nothing, because every one of those is guarded by the object's own
//! etag (`If-Match`) and installs no citation. Between barriers nobody
//! holds the cell, so a second writer on the same workspace is Ready in
//! checkout time and publishes every floor — the life-long lease this
//! replaces made it wait the first writer's LIFETIME in
//! ContainerCreating (`verbs.rs` used to call it "a DEADLOCK dressed as
//! mutual exclusion", and had already taken checkout out of it).
//!
//! Deliberately NOT `tier::epoch::claim`: that path (a) runs the
//! bucket-wide MPU takeover sweep, which a project-scoped proxy denies
//! (the sweep is an operator-side job in lean — plan §2.4), and (b)
//! gates self-recognition on the hub's state-directory occupancy lock.
//! Lean's self-recognition token is the PERSISTED INCARNATION ID: it is
//! emptyDir-scoped, so only the same pod's restarted container inherits
//! it — which is exactly the one case where immediate self-supersede is
//! safe. A replacement pod gets a fresh id and must wait out the quiet
//! polls; that observation ({last_token, quiet_polls}) persists so a
//! container restart RESUMES it instead of resetting the clock.
//!
//! The claim step is one read plus at most one CAS. Its verdicts:
//!
//! - a FRESH cell: acquire.
//! - a cell that names THIS incarnation at THIS epoch: our own acquire
//!   whose response was lost — adopt it.
//! - a RELEASED cell reserved for nobody, or for us: acquire (a clean
//!   handoff; no manifest rotation).
//! - a RELEASED cell reserved for someone else: queue up once, and wait
//!   — unless the reservation has stood unclaimed across
//!   `HANDOFF_QUIET_POLLS` spaced observations (the reserved holder
//!   died), in which case take it.
//! - a cell HELD by someone else (or by us at an epoch we never
//!   recorded: an orphaned acquire): queue up once, and wait — unless it
//!   has stood still across `QUIET_POLLS` spaced observations (the
//!   holder died inside its commit section), in which case DEPOSE it and
//!   rotate the manifest, exactly the straggler fence there always was.
//!
//! An observation is "spaced" when at least `QUIET_SPACING_SECS` passed
//! since the previous counted one. The wait loop polls faster than
//! that (`CLAIM_POLL_SECS`), so a release is seen within a second, but
//! deadness is still judged over a minute of a token that does not move.
//! A waiter's own enqueue moves the token; the holder's next renew or
//! handoff 412s, re-reads, and adopts the longer queue (`renew`,
//! `release`).

use std::sync::Arc;
use std::time::{Duration, Instant};

use flint_store::{EpochLease, EpochState, ObjectStore, StoreError};

use super::state::Incarnation;
use super::{manifest, LeanError, LeanResult, Syncer};

/// Spaced quiet observations before a HELD cell's holder is judged dead
/// and deposed (6 x 10 s: the takeover threshold the drills were run at).
pub const QUIET_POLLS: u32 = 6;
/// Spaced quiet observations before a RELEASED cell's reservation is
/// judged abandoned (the reserved waiter died before claiming).
pub const HANDOFF_QUIET_POLLS: u32 = 2;
/// Minimum spacing between two observations that COUNT toward either
/// threshold. Polls in between only look for a release or a handoff.
pub const QUIET_SPACING_SECS: u64 = 10;
/// The wait loop's poll cadence: one GET of a few-hundred-byte cell.
pub const CLAIM_POLL_SECS: u64 = 1;
/// The poll cadence of the waiter at the HEAD of the queue: it is the one
/// the holder's handoff reserves the cell for, and every millisecond it
/// takes to notice is a millisecond the cell stands idle with the rest of
/// the queue behind it. In the writers drill (6 writers, 1 s polls) the
/// cell sat released a median 614 ms before the next holder claimed it,
/// on top of a 973 ms hold. Deadness is judged on `QUIET_SPACING_SECS`,
/// not on this, so the thresholds do not move.
pub const CLAIM_POLL_HEAD_MS: u64 = 200;
/// Longest a barrier waits for the cell before failing (and being
/// retried at the next floor). Comfortably past the deposal threshold
/// plus the spacing it needs, so a dead holder is always deposed within
/// one wait; a LIVE holder that never releases is a bug, and the
/// deadline is what keeps it from being a hang.
pub const CLAIM_DEADLINE_SECS: u64 = 150;

pub enum ClaimOutcome {
    /// Fresh, released-for-us, deposed, or adopted: held.
    Claimed(EpochLease),
    /// Not ours yet. `quiet_polls` is the spaced-observation count
    /// against the current verdict's threshold; `behind` names the
    /// holder or the reserved waiter we are waiting on; `next` says this
    /// waiter heads the queue behind a held cell, so the holder's handoff
    /// will name it.
    Waiting { quiet_polls: u32, behind: Option<String>, next: bool },
}

/// Say so, loudly, if another product also writes this prefix.
///
/// Detection, deliberately NOT enforcement — see the forge twin. A
/// lean workspace and a forge repository on one prefix arbitrate on
/// different cells and will never fence each other, so without this
/// nothing anywhere reports the condition.
///
/// Note what is NOT a finding: forge's legible export IS a lean
/// workspace, published by forge rather than by an agent's syncer, so
/// a syncer reading an export prefix finds only its own kind of cell.
/// And an export nested under a repository's prefix puts this cell at
/// `<prefix>/inner/.flint/lean/epoch`, which is not the key forge
/// probes — the exact-key probe is what keeps that from reading as a
/// collision.
pub async fn warn_if_prefix_is_shared(sc: &Syncer) {
    // A published mirror does not probe. Two reasons, and the second is
    // the load-bearing one:
    //
    // Forge spawns a barrier per export, so this would be a recurring
    // read for a condition that cannot change between two barriers of
    // the same publisher. And forge echoes only the child lines
    // containing "barrier" (export.rs:389), so a warning raised here
    // would be read by nobody — a check whose output is discarded is
    // worse than no check, because it looks like coverage.
    //
    // The prefix is still probed, ONCE, by the publisher at startup:
    // see `warn_if_export_prefix_is_shared` in forge's twin. Moved, not
    // dropped — an export prefix is a prefix two products both write
    // legitimately, which makes it likelier than most to have a third
    // pointed at it.
    if sc.cfg.sole_writer {
        return;
    }
    let found = flint_store::layout::neighbours(
        sc.store.as_ref(),
        &sc.cfg.prefix,
        flint_store::layout::Writer::LeanWorkspace,
    )
    .await
    .unwrap_or_default();
    for f in found {
        eprintln!("flint-sync: {}", f.report());
    }
}

/// This pod's identity, minted once and persisted in the state
/// directory (emptyDir-scoped: a restarted container inherits it, a
/// replacement pod does not).
pub fn incarnation(sc: &Syncer) -> LeanResult<Incarnation> {
    match sc.state.load_incarnation()? {
        Some(i) => Ok(i),
        None => {
            let i = Incarnation {
                holder_id: format!("lean-{}", uuid::Uuid::new_v4()),
                epoch: 0,
                last_token: None,
                quiet_polls: 0,
            };
            sc.state.save_incarnation(&i)?;
            Ok(i)
        }
    }
}

/// One claim step. The caller loops on `Waiting` at its poll cadence;
/// each call performs one read and at most one CAS (an acquire, or an
/// enqueue). `count` says whether this observation is spaced far
/// enough from the last counted one to advance the quiet thresholds —
/// the loop decides that from its own clock, so the judgement of
/// deadness stays tied to wall time however fast the loop polls.
pub async fn claim_step(sc: &mut Syncer, count: bool) -> LeanResult<ClaimOutcome> {
    let store: &Arc<dyn ObjectStore> = &sc.store;
    let key = sc.cfg.epoch_key();
    let mut inc = incarnation(sc)?;

    let observed = store.epoch_read(&key).await?;
    let Some(state) = observed else {
        return match store.epoch_acquire(&key, &inc.holder_id, None).await {
            Ok(lease) => {
                inc.epoch = lease.epoch;
                inc.last_token = None;
                inc.quiet_polls = 0;
                sc.state.save_incarnation(&inc)?;
                sc.lease = Some(lease.clone());
                sc.cell_written_at = Some(std::time::Instant::now());
                sc.trace("claim", serde_json::json!({"verdict": "claimed", "how": "fresh", "epoch": lease.epoch}));
                Ok(ClaimOutcome::Claimed(lease))
            }
            Err(StoreError::PreconditionFailed(_)) | Err(StoreError::Conflict(_)) => {
                Ok(ClaimOutcome::Waiting { quiet_polls: 0, behind: None, next: false })
            }
            Err(e) => Err(e.into()),
        };
    };

    // Self-recognition needs the EPOCH too (review 2026-09-12, lease-3 /
    // audit #7): a cell naming this holder at an epoch this incarnation
    // never recorded is our own acquire whose response was lost, or
    // whose rotation failed after it landed — the straggler it deposed
    // may still be mid-barrier, so that path takes the takeover
    // rotation like any other.
    let same_holder = state.holder_id == inc.holder_id;
    let ours = same_holder && state.epoch == inc.epoch && !state.released;
    let quiet = inc.last_token.as_deref() == Some(state.token.as_str());
    let counted_quiet = if count {
        if quiet {
            inc.quiet_polls + 1
        } else {
            0
        }
    } else {
        inc.quiet_polls
    };

    if ours {
        // Our own acquire whose response was lost, or a cell a previous
        // container of this pod is still named on at the epoch we
        // recorded: adopt it IN PLACE — token and queue as they stand,
        // no write. (Re-acquiring would move the epoch for nothing and
        // make every reader of the cell count a barrier that never ran.)
        let lease = EpochLease {
            holder_id: state.holder_id,
            epoch: state.epoch,
            token: state.token,
            waiters: state.waiters,
        };
        inc.last_token = None;
        inc.quiet_polls = 0;
        sc.state.save_incarnation(&inc)?;
        sc.lease = Some(lease.clone());
        // Adopted, not written: the commit section's first delete renews
        // before it trusts the token (review 2026-09-18, C2 and L8).
        sc.cell_written_at = None;
        sc.trace("claim", serde_json::json!({"verdict": "claimed", "how": "adopted-own", "epoch": lease.epoch}));
        return Ok(ClaimOutcome::Claimed(lease));
    }

    // The verdict. `take` = acquire now; `rotate` = the acquire is a
    // deposal of a possibly-live straggler and must fence it.
    let (take, rotate, behind) = if same_holder && !state.released {
        // Orphaned own: our acquire landed and its response was lost, or
        // its rotation failed after it landed. The straggler it deposed
        // may still be mid-commit, so this takes the rotation too.
        (true, true, None)
    } else if state.released {
        match state.handoff.as_deref() {
            None => (true, false, None),
            Some(h) if h == inc.holder_id => (true, false, None),
            Some(h) => (counted_quiet >= HANDOFF_QUIET_POLLS, false, Some(h.to_string())),
        }
    } else {
        // Held by someone else, or by us at an epoch we never recorded.
        (counted_quiet >= QUIET_POLLS, true, Some(state.holder_id.clone()))
    };

    if take {
        // Rotation is needed ONLY for the unreleased-foreign takeover (a
        // possibly-live straggler mid-commit). A released cell is a
        // clean handoff — the holder's barrier completed before its
        // release — and self-recognition means the previous container's
        // process (and any in-flight write of its) died with it.
        // Rotating on those paths is pure manifest churn: at 100k+
        // entries it is a multi-MB GET+PUT per claim, it double-bumps
        // seq, and it defeats the no-change barrier's early exit
        // (measured on the 0b rig). Under the per-barrier lease that
        // would be every boundary.
        let prior = serde_json::json!({"holder": state.holder_id, "epoch": state.epoch, "released": state.released,
            "handoff": state.handoff, "waiters": state.waiters});
        let how = if rotate && same_holder {
            "orphaned-own"
        } else if rotate {
            "deposed"
        } else if state.handoff.as_deref().map(|h| h != inc.holder_id).unwrap_or(false) {
            "skipped-handoff"
        } else {
            "released"
        };
        return match store.epoch_acquire(&key, &inc.holder_id, Some(&state)).await {
            Ok(lease) => {
                if rotate {
                    manifest::rotate_for_takeover(store.as_ref(), &sc.cfg, lease.epoch).await?;
                }
                inc.epoch = lease.epoch;
                inc.last_token = None;
                inc.quiet_polls = 0;
                sc.state.save_incarnation(&inc)?;
                sc.lease = Some(lease.clone());
                sc.cell_written_at = Some(std::time::Instant::now());
                sc.trace("claim", serde_json::json!({"verdict": "claimed", "how": how, "epoch": lease.epoch, "prior": prior}));
                Ok(ClaimOutcome::Claimed(lease))
            }
            Err(StoreError::PreconditionFailed(_)) | Err(StoreError::Conflict(_)) => {
                // Lost the race (a rival acquire, or an enqueue moved the
                // token); restart the observation.
                inc.last_token = None;
                inc.quiet_polls = 0;
                sc.state.save_incarnation(&inc)?;
                Ok(ClaimOutcome::Waiting { quiet_polls: 0, behind, next: false })
            }
            Err(e) => Err(e.into()),
        };
    }

    // Not ours yet: make sure we are in the queue, ONCE. The enqueue
    // moves the token, so the observation restarts from the token we
    // wrote — our own append must not read as the holder's renewal,
    // and it must not reset a count that a rival's append did not.
    let token = if state.waiters.iter().any(|w| w == &inc.holder_id) {
        state.token.clone()
    } else {
        match store.epoch_enqueue(&key, &state, &inc.holder_id).await {
            Ok(after) => after.token,
            Err(StoreError::PreconditionFailed(_)) | Err(StoreError::Conflict(_)) => {
                // Somebody else moved the cell first; observe again.
                state.token.clone()
            }
            Err(e) => return Err(e.into()),
        }
    };
    if count {
        inc.quiet_polls = counted_quiet;
        inc.last_token = Some(token);
        sc.state.save_incarnation(&inc)?;
    }
    // First in the queue behind a HELD cell (our enqueue appended us to
    // whatever `state.waiters` held, or we were already in it).
    let next = !state.released
        && state.waiters.iter().find(|w| **w != state.holder_id).map(|w| w == &inc.holder_id).unwrap_or(true);
    Ok(ClaimOutcome::Waiting { quiet_polls: counted_quiet, behind, next })
}

/// The wait loop: claim the cell for a commit section, or give up
/// after `CLAIM_DEADLINE_SECS` so a holder that never releases turns
/// into a failed (and retried) barrier rather than a hang.
pub async fn claim(sc: &mut Syncer) -> LeanResult<EpochLease> {
    let started = Instant::now();
    let mut last_counted: Option<Instant> = None;
    let mut reported: Option<String> = None;
    let spacing = Duration::from_secs(sc.cfg.claim_quiet_spacing_secs);
    let deadline = Duration::from_secs(sc.cfg.claim_deadline_secs);
    let poll = Duration::from_secs(sc.cfg.claim_poll_secs);
    let head_poll = Duration::from_millis(sc.cfg.claim_poll_head_ms).min(poll);
    loop {
        let count = last_counted.map(|t| t.elapsed() >= spacing).unwrap_or(true);
        match claim_step(sc, count).await? {
            ClaimOutcome::Claimed(lease) => return Ok(lease),
            ClaimOutcome::Waiting { quiet_polls, behind, next } => {
                if count {
                    last_counted = Some(Instant::now());
                    sc.trace("claim", serde_json::json!({"verdict": "waiting", "behind": behind, "quiet_polls": quiet_polls,
                        "waited_ms": started.elapsed().as_millis() as u64}));
                    // Once per spaced observation, not once per second.
                    let line = format!(
                        "flint-sync: waiting for the publish fence behind {} (quiet {quiet_polls})",
                        behind.as_deref().unwrap_or("a rival claim")
                    );
                    if reported.as_deref() != Some(line.as_str()) {
                        eprintln!("{line}");
                        reported = Some(line);
                    }
                }
                if started.elapsed() >= deadline {
                    sc.trace("claim", serde_json::json!({"verdict": "deadline", "behind": behind,
                        "waited_ms": started.elapsed().as_millis() as u64}));
                    return Err(LeanError::State(format!(
                        "could not acquire the publish fence within {}s \
                         (behind {}); this barrier is abandoned and retried at the next floor — \
                         its uploads are durable and the next barrier adopts them",
                        deadline.as_secs(),
                        behind.as_deref().unwrap_or("a rival claim")
                    )));
                }
                tokio::time::sleep(if next { head_poll } else { poll }).await;
            }
        }
    }
}

/// What this syncer is OBSERVED to be doing, for the cell's echo
/// (boundary-verbs plan §2.6). Computed from local files only — the
/// same store-free discipline as `write_gauges`, and for the same
/// reason: this rides writes that already happen, so it must not add a
/// request to the one tick every idle workspace in the fleet pays (leg
/// B8's oracle counts them).
fn observed_echo(sc: &Syncer) -> Option<String> {
    let g = sc.load_gauges().ok()?;
    let (seq, unix) = g.last_boundary.as_ref().map(|b| (b.seq, b.unix)).unwrap_or((0, 0));
    // The pre-flight verdict lives in the marker the AGENT reads; the
    // operator has no other way to see it, so it rides the echo.
    let caps: Option<super::control::Capabilities> =
        std::fs::read(sc.cfg.control_dir().join(super::control::CAPABILITIES))
            .ok()
            .and_then(|b| serde_json::from_slice(&b).ok());
    serde_json::to_string(&flint_store::LeaseEcho {
        syncer_version: super::SYNCER_VERSION.to_string(),
        protocol: super::SENTINEL_PROTOCOL,
        last_cited_seq: seq,
        last_cited_unix: unix,
        sentinel_verbs_active: caps.map(|c| !c.verbs.is_empty()).unwrap_or(false),
        metrics_bound: sc.load_metrics_posture().filter(|m| m.enabled).map(|m| m.bound),
    })
    .ok()
}

/// Re-read the cell after a 412 and, if it still names this holder at
/// this epoch unreleased, hand back its current token and queue: the
/// only things that move OUR cell's token are our own writes whose
/// response was lost and the waiters' enqueues, and neither is a
/// takeover. Anything else is the fence.
async fn still_ours(sc: &Syncer, key: &str, lease: &EpochLease) -> Option<EpochLease> {
    match sc.store.epoch_read(key).await {
        Ok(Some(state))
            if state.holder_id == lease.holder_id && state.epoch == lease.epoch && !state.released =>
        {
            Some(EpochLease {
                holder_id: state.holder_id,
                epoch: state.epoch,
                token: state.token,
                waiters: state.waiters,
            })
        }
        _ => None,
    }
}

/// Renew the held lease inside a commit section; a 412 that is not our
/// own moved token means deposed — the caller must abandon the barrier
/// (self-fence).
///
/// The renewal also carries the observed-state echo (§2.6).
pub async fn renew(sc: &mut Syncer) -> LeanResult<()> {
    let key = sc.cfg.epoch_key();
    let lease = sc
        .lease
        .clone()
        .ok_or_else(|| LeanError::State("renew without a lease".into()))?;
    let echo = observed_echo(sc);
    match sc.store.epoch_renew(&key, &lease, echo.as_deref()).await {
        Ok(l) => {
            sc.lease = Some(l);
            sc.cell_written_at = Some(std::time::Instant::now());
            // The renewal is a probe of our own credentials, so it is
            // also a place a pause can be observed to have ENDED.
            // Best-effort: a gauge that failed to write must not fail a
            // renewal that succeeded.
            let _ = sc.clear_auth_pause();
            Ok(())
        }
        Err(StoreError::PreconditionFailed(e)) => {
            // A 412 is not yet a deposal. The renew CAS is If-Match on
            // OUR token, and three things move that token: a
            // successor's acquire, a waiter's enqueue, or our own
            // previous write whose RESPONSE was lost. Only the first is
            // a takeover, and treating the others as one made a live
            // holder fence itself (audit 2026-09-03, finding 2). One
            // read tells the cases apart.
            match still_ours(sc, &key, &lease).await {
                Some(adopted) => {
                    eprintln!(
                        "flint-sync: renew 412 on a cell that is still ours (epoch {}, {} queued): \
                         adopting its token, not fencing",
                        adopted.epoch,
                        adopted.waiters.len()
                    );
                    // Review 2026-09-12, lease-2: the adoption alone
                    // writes nothing, so the token would stand still
                    // for a whole takeover threshold and a waiter could
                    // count a live holder dead. One renew moves it; if
                    // that write fails the adopted token stands and the
                    // next tick tries again.
                    match sc.store.epoch_renew(&key, &adopted, echo.as_deref()).await {
                        Ok(l) => {
                            sc.lease = Some(l);
                            sc.cell_written_at = Some(std::time::Instant::now());
                        }
                        Err(_) => sc.lease = Some(adopted),
                    }
                    let _ = sc.clear_auth_pause();
                    Ok(())
                }
                None => {
                    sc.lease = None;
                    Err(LeanError::Fenced(format!("deposed at renew: {e}")))
                }
            }
        }
        // 401/403 is not contention and not a bucket fault; retrying
        // cannot fix it. Record when the pause began while we still can:
        // nothing we write to the STORE can carry this, because the
        // request that would carry it is the one being refused (design
        // §6.3).
        Err(e @ StoreError::Auth(_)) => {
            let _ = sc.note_auth_pause();
            Err(e.into())
        }
        Err(e) => Err(e.into()),
    }
}

/// Renew if the cell has not moved for half the takeover threshold
/// (`QUIET_POLLS` spaced observations: 30 s of 60 at the shipped spacing),
/// so a waiter never counts a live holder dead however long its commit
/// section runs (review 2026-09-18, M3) — judged by THIS node's
/// monotonic clock (`cell_written_at`, stamped at every claim and renew):
/// a node clock behind the store's never mistakes a stale token for a
/// fresh one (review 2026-09-18, M4 / lease-4). Only a deposal is an
/// error; any other failure leaves the token where it was, the next call
/// retries, and the section goes on — a renew is availability, and the
/// fence before the CAS (`verify_not_deposed`) is what safety rests on.
pub async fn renew_if_due(sc: &mut Syncer) -> LeanResult<()> {
    if sc.lease.is_none() {
        return Ok(());
    }
    let after = Duration::from_secs(QUIET_POLLS as u64 * sc.cfg.claim_quiet_spacing_secs / 2);
    let due = sc.cell_written_at.map_or(true, |t| t.elapsed() >= after);
    if !due {
        return Ok(());
    }
    match renew(sc).await {
        Ok(()) => {
            sc.trace("renew", serde_json::json!({"epoch": sc.lease.as_ref().map(|l| l.epoch)}));
            Ok(())
        }
        Err(e @ LeanError::Fenced(_)) => {
            sc.trace("fence", serde_json::json!({"where": "renew", "detail": e.to_string()}));
            Err(e)
        }
        Err(e) => {
            eprintln!("flint-sync: renew inside the commit section failed ({e}); the next one retries");
            Ok(())
        }
    }
}

/// The claim precondition (audit 2026-09-03, finding 5). With a project
/// id stamped (`FLINT_SYNC_PROJECT_ID`), refuse to run over a prefix
/// whose claim cell names ANOTHER project. The operator's refuse-foreign
/// was advisory on the data plane: a CR the operator had not yet judged
/// resolved to its spec, and the syncer checked out and republished over
/// the foreign project's manifest. The cell is durable in the bucket, so
/// the data plane can read it — one GET before the first claim step.
/// Absent cell ⇒ a fresh prefix, claimable; unstamped ⇒ no check (the
/// pre-operator posture, which the drill runs). A foreign claim is
/// `LeanError::Refused`: the process exits `EXIT_REFUSED`, which the
/// delivery treats as final (tear down, name the reason) rather than
/// as one more crash to restart.
pub async fn verify_claim(sc: &Syncer) -> LeanResult<()> {
    let Some(mine) = sc.cfg.project_id.as_deref() else { return Ok(()) };
    let key = sc.cfg.claim_key();
    match sc.store.get_whole(&key, None).await {
        Ok((_, body)) => {
            let doc: serde_json::Value = serde_json::from_slice(&body)
                .map_err(|e| LeanError::State(format!("claim cell {key} is unparseable: {e}")))?;
            match doc.get("project_id").and_then(|v| v.as_str()) {
                Some(p) if p == mine => Ok(()),
                Some(p) => Err(LeanError::Refused(format!(
                    "prefix {} is claimed by project {p:?}; this workspace is project {mine:?} — \
                     refusing to check out or publish over another project's data (delete the \
                     standing claim explicitly if this reuse is intended)",
                    sc.cfg.prefix
                ))),
                None => Err(LeanError::State(format!("claim cell {key} names no project_id"))),
            }
        }
        Err(StoreError::NotFound(_)) => Ok(()),
        Err(e) => Err(e.into()),
    }
}

/// Release at the end of a commit section: the cell is handed to the
/// queue head and keeps this barrier's echo.
///
/// Every waiter's enqueue moves the token, so under contention the handoff
/// can 412 several times in a row, and S3 answers a conditional write that
/// races another with 409 ConditionalRequestConflict. Each is re-read: while
/// the cell still names this holder at this epoch, unreleased, the handoff
/// is retried on the adopted token. Only a cell that is no longer ours ends
/// the attempt quietly. Returning with the cell still held — what a second
/// 412 or a 409 used to do — leaves nobody to release it: the queue waits
/// out the deposal threshold and the takeover rotates the manifest (the
/// contention drill, 2026-09-14).
pub async fn release(sc: &mut Syncer) -> LeanResult<()> {
    let key = sc.cfg.epoch_key();
    let Some(lease) = sc.lease.take() else { return Ok(()) };
    let echo = observed_echo(sc);
    sc.trace("release", serde_json::json!({"epoch": lease.epoch, "waiters_at_claim": lease.waiters}));
    // `release` above is stamped before the request, `handed_off` after it
    // lands: the gap between one holder's release and the next claim is
    // the handoff's own latency plus the reserved waiter's poll delay.
    let mut current = lease;
    for attempt in 0..RELEASE_ATTEMPTS {
        match sc.store.epoch_handoff(&key, &current, echo.as_deref()).await {
            Ok(()) => {
                sc.trace("handed_off", serde_json::json!({"epoch": current.epoch, "retried": attempt > 0, "attempts": attempt + 1}));
                return Ok(());
            }
            Err(StoreError::PreconditionFailed(_)) | Err(StoreError::Conflict(_)) => {
                if attempt > 0 {
                    // Waiters enqueue in bursts; step out of the burst.
                    tokio::time::sleep(Duration::from_millis(20 * attempt as u64)).await;
                }
                match sc.store.epoch_read(&key).await {
                    Ok(Some(state))
                        if state.holder_id == current.holder_id
                            && state.epoch == current.epoch
                            && !state.released =>
                    {
                        current = EpochLease {
                            holder_id: state.holder_id,
                            epoch: state.epoch,
                            token: state.token,
                            waiters: state.waiters,
                        };
                    }
                    // Released, rotated or taken: the cell is the successor's.
                    Ok(_) => return Ok(()),
                    Err(e) if attempt + 1 < RELEASE_ATTEMPTS => {
                        eprintln!("flint-sync: re-reading the publish fence for its handoff: {e}; retrying");
                    }
                    Err(e) => return Err(e.into()),
                }
            }
            Err(e) => return Err(e.into()),
        }
    }
    Err(LeanError::State(format!(
        "the publish fence handoff lost {RELEASE_ATTEMPTS} races to the queue; the cell stays ours \
         until this writer's next claim adopts it, or a waiter deposes it"
    )))
}

/// Handoff attempts before a release gives up (each lost to a waiter's
/// enqueue or a racing conditional write, re-read between them).
pub const RELEASE_ATTEMPTS: u32 = 8;

/// A restarted container that finds the cell HELD by its own
/// incarnation releases it: it holds nothing in memory, the intent
/// journal replays whatever the previous container left, and a cell
/// left held would cost every other writer a 60 s deposal wait.
pub async fn release_stale_own(sc: &mut Syncer) -> LeanResult<()> {
    let key = sc.cfg.epoch_key();
    let inc = incarnation(sc)?;
    if let Some(state) = sc.store.epoch_read(&key).await? {
        if state.holder_id == inc.holder_id && !state.released {
            eprintln!(
                "flint-sync: the publish fence was left held by a previous container of this pod \
                 (epoch {}); releasing it",
                state.epoch
            );
            // Review 2026-09-18, H2: an epoch this incarnation never
            // recorded is an acquire whose container died before its
            // takeover rotation — `claim_step`'s orphaned-own arm, which a
            // restart never reaches. The holder it deposed may still be
            // mid-commit, and neither this release nor the next claim of
            // a released cell rotates, so its CAS on the unmoved pointer
            // landed. Rotate first, as that arm does.
            let rotated = state.epoch != inc.epoch;
            if rotated {
                manifest::rotate_for_takeover(sc.store.as_ref(), &sc.cfg, state.epoch).await?;
            }
            // The model's Restart: a new incarnation of this writer, which
            // drops the barrier a previous one left in flight and frees the
            // cell it held.
            sc.trace("start", serde_json::json!({"held": true, "rotated": rotated}));
            sc.lease = Some(EpochLease {
                holder_id: state.holder_id,
                epoch: state.epoch,
                token: state.token,
                waiters: state.waiters,
            });
            return release(sc).await;
        }
    }
    sc.trace("start", serde_json::json!({"held": false, "rotated": false}));
    Ok(())
}

// An observed cell, re-exported for the callers that fence on one.
pub type Observed = EpochState;