Skip to content

perf(turns): long-lived authority + remove redundant global commit_gate from the turn-state row store (#6263 Step 1) #6281

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ilblackdragon merged 2 commits into from
Jul 19, 2026
Merged

perf(turns): long-lived authority + remove redundant global commit_gate from the turn-state row store (#6263 Step 1) #6281

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d588a37
perf(turns): long-lived authority reuse + remove redundant global com…
ilblackdragon Jul 19, 2026
4faac10
fix(turns): invalidate the row-store snapshot cache on a failed put_l…
ilblackdragon Jul 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 57 additions & 36 deletions crates/ironclaw_turns/src/filesystem_store/row_store.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,9 @@ const ROW_COLLECTION_READ_CONCURRENCY: usize = 32;
/// Transitions still delegate to [`InMemoryTurnStateStore`]; only the durable
/// representation changes from whole-snapshot CAS to a typed append log plus a
/// process-local hot snapshot cache.
// arch-exempt: large_file, long-lived embedded-authority refactor adds a shared
// overlay-store acquisition helper (deduplicating the apply / targeted-delta
// paths) rather than a second store-acquisition pipeline, plan #6263
pub struct FilesystemTurnStateRowStore<F>
where
F: RootFilesystem,
Expand All @@ -70,7 +73,6 @@ where
limits: InMemoryTurnStateStoreLimits,
admission_limit_provider: Arc<dyn TurnAdmissionLimitProvider>,
snapshot_state: AsyncMutex<Option<RowSnapshotState>>,
commit_gate: AsyncMutex<()>,
legacy_migration_gate: Arc<AsyncMutex<()>>,
materialize_gate: Arc<AsyncMutex<()>>,
runner_leases: RunnerLeaseMemory,
Expand Down Expand Up @@ -146,7 +148,6 @@ where
limits: InMemoryTurnStateStoreLimits::default(),
admission_limit_provider: Arc::new(AllowAllTurnAdmissionLimitProvider),
snapshot_state: AsyncMutex::new(None),
commit_gate: AsyncMutex::new(()),
legacy_migration_gate: Arc::new(AsyncMutex::new(())),
materialize_gate: Arc::clone(&materialize_gate),
runner_leases: Arc::new(RwLock::new(HashMap::new())),
Expand Down Expand Up @@ -746,6 +747,37 @@ where
)
}

/// Acquire the long-lived embedded authority with the runner-lease overlay
/// applied, without rebuilding it from a full-snapshot clone.
///
/// For [`RunnerLeaseOverlay::None`] and [`RunnerLeaseOverlay::Run`] this
/// reuses the cached authority in place (the `Run` case overlays a single
/// run record's live lease heartbeat onto it) — an O(1) reuse rather than a
/// per-op O(store-size) `from_persistence_snapshot` rebuild. Only
/// [`RunnerLeaseOverlay::All`] (expired-lease recovery) still rebuilds from
/// an overlaid snapshot clone, because it must overlay every run's lease.
async fn acquire_overlaid_store(
&self,
cached_store: Arc<InMemoryTurnStateStore>,
overlay_baseline: Option<TurnPersistenceSnapshot>,
overlay_run: Option<TurnRunRecord>,
overlay: RunnerLeaseOverlay,
) -> Result<Arc<InMemoryTurnStateStore>, TurnError> {
if let Some(baseline) = overlay_baseline {
let (overlaid_snapshot, _) = self
.runner_lease_store()
.overlay((baseline, None), overlay)
.await?;
Ok(Arc::new(self.build_in_memory_store(overlaid_snapshot)?))
} else {
if let Some(run) = overlay_run {
let overlaid = self.runner_lease_store().overlay_run_record(run).await?;
cached_store.overlay_runner_lease_record(overlaid)?;
}
Ok(cached_store)
}
}

async fn apply<T, A, Fut>(
&self,
overlay: RunnerLeaseOverlay,
Expand All @@ -757,20 +789,31 @@ where
T: Send,
{
let critical = async {
let _commit_guard = self.commit_gate.lock().await;
let mut guard = self.snapshot_state.lock().await;
let mut refreshed_after_stale_error = false;
loop {
self.ensure_snapshot_cache_for_mutation(&mut guard).await?;
let (baseline, current_journal_seq) = guard
.as_ref()
.map(|state| (state.snapshot.clone(), state.journal_seq))
.unwrap_or_else(|| (TurnPersistenceSnapshot::default(), SeqNo::ZERO));
let (overlaid_snapshot, _) = self
.runner_lease_store()
.overlay((baseline.clone(), None), overlay)
let (baseline, current_journal_seq, cached_store, overlay_baseline, overlay_run) = {
let state = guard.as_ref().ok_or_else(|| TurnError::Unavailable {
reason: "row snapshot cache was not initialized".to_string(),
})?;
let overlay_baseline =
matches!(overlay, RunnerLeaseOverlay::All).then(|| state.snapshot.clone());
let overlay_run = match overlay {
RunnerLeaseOverlay::Run(run_id) => state.run_record_by_id(run_id),
RunnerLeaseOverlay::None | RunnerLeaseOverlay::All => None,
};
(
state.snapshot.clone(),
state.journal_seq,
Arc::clone(&state.store),
overlay_baseline,
overlay_run,
)
};
let store = self
.acquire_overlaid_store(cached_store, overlay_baseline, overlay_run, overlay)
.await?;
let store = Arc::new(self.build_in_memory_store(overlaid_snapshot)?);
let outcome = apply(Arc::clone(&store)).await;
let mut new_snapshot = store.persistence_snapshot();
preserve_loop_checkpoints(&baseline, &mut new_snapshot);
Expand Down Expand Up @@ -1567,17 +1610,6 @@ where
}
}

async fn apply_cached_delta(&self, delta: SnapshotDelta) -> Result<(), TurnError> {
if delta.is_empty() {
return Ok(());
}
let mut guard = self.snapshot_state.lock().await;
if let Some(state) = guard.as_mut() {
state.apply_delta(delta, state.journal_seq)?;
}
Ok(())
}

async fn apply_with_targeted_delta<T, A, Fut, D>(
&self,
overlay: RunnerLeaseOverlay,
Expand All @@ -1597,7 +1629,6 @@ where
T: Send,
{
let critical = async {
let _commit_guard = self.commit_gate.lock().await;
let mut guard = self.snapshot_state.lock().await;
let mut build_delta = Some(build_delta);
let mut refreshed_after_stale_error = false;
Expand All @@ -1620,19 +1651,9 @@ where
overlay_run,
)
};
let store = if let Some(baseline) = overlay_baseline {
let (overlaid_snapshot, _) = self
.runner_lease_store()
.overlay((baseline, None), overlay)
.await?;
Arc::new(self.build_in_memory_store(overlaid_snapshot)?)
} else {
if let Some(run) = overlay_run {
let overlaid = self.runner_lease_store().overlay_run_record(run).await?;
cached_store.overlay_runner_lease_record(overlaid)?;
}
cached_store
};
let store = self
.acquire_overlaid_store(cached_store, overlay_baseline, overlay_run, overlay)
.await?;
let outcome = apply(Arc::clone(&store)).await;
let value = match outcome {
Ok(value) => value,
Expand Down
18 changes: 16 additions & 2 deletions crates/ironclaw_turns/src/filesystem_store/row_store/traits.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -365,13 +365,27 @@ where
let record = loop_checkpoint_record_from_request(request);
let delta = SnapshotDelta::default().set_loop_checkpoints_upsert(vec![record.clone()]);
let ack = {
let _commit_guard = self.commit_gate.lock().await;
// Serialize the durable enqueue on the shared snapshot-state lock
// — the same lock every apply path holds across its
// read-seq -> enqueue window — so the delta journal observes a
// consistent append order without a separate commit gate.
let mut guard = self.snapshot_state.lock().await;
let ack = self
.enqueue_delta(row_store_durable_delta(delta.clone()))
.map_err(|error| match error {
RowPersistError::Turn(error) => error,
})?;
self.apply_cached_delta(delta).await?;
if let Some(state) = guard.as_mut()
&& let Err(error) = state.apply_delta(delta, state.journal_seq)
{
// A failed in-memory apply may leave the cached snapshot
// partially mutated; drop it so the next read rebuilds from
// durable state, matching `apply` / `apply_with_targeted_delta`.
// (The durable enqueue already succeeded, so recovery replays
// it — the cache is the only thing to invalidate.)
*guard = None;
return Err(error);
}
ack
};
self.await_pending_commit(
Expand Down
Loading