Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
26 changes: 17 additions & 9 deletions crates/tracedecay-daemon-service/src/invocation/types.rs
Original file line number Diff line number Diff line change
Expand Up @@ -207,6 +207,7 @@ impl BoundedHookOrchestratorV1 {
address: &HookOrchestrationAddressV1,
operation: &Arc<HookOrchestrationInFlightEntryV1>,
emit_terminal: bool,
permit: Option<tokio::sync::OwnedSemaphorePermit>,
) {
let completions = {
let mut in_flight = in_flight
Expand All @@ -233,6 +234,12 @@ impl BoundedHookOrchestratorV1 {
{
in_flight.events.remove(&operation.event);
}
// Admission checks the in-flight maps and takes a permit under
// this lock, so the cycle leaves both in one step: no admission
// sees it gone yet still holding its permit, or sees its permit
// free while it still looks in flight. Its terminals, the host's
// cue to send the session's next boundary, follow.
drop(permit);
completions
};
if emit_terminal {
Expand Down Expand Up @@ -388,15 +395,14 @@ impl BoundedHookOrchestratorV1 {
let superseded = operation
.superseded
.load(std::sync::atomic::Ordering::Acquire);
Self::settle_operation(&in_flight, &address, &operation, superseded);
Self::settle_operation(&in_flight, &address, &operation, superseded, None);
return;
};
if cancellation.is_cancelled() || work_cancellation.is_cancelled() {
let superseded = operation
.superseded
.load(std::sync::atomic::Ordering::Acquire);
drop(permit);
Self::settle_operation(&in_flight, &address, &operation, superseded);
Self::settle_operation(&in_flight, &address, &operation, superseded, Some(permit));
return;
}
let resumable = request.clone();
Expand Down Expand Up @@ -456,19 +462,21 @@ impl BoundedHookOrchestratorV1 {
.load(std::sync::atomic::Ordering::Acquire)
}
};
// Free the permit before the operation leaves the in-flight maps.
// An admission that no longer finds this operation must not be
// backpressured by it, and a completion fires only once the next
// admission can run.
drop(permit);
Self::settle_operation(&in_flight, &address, &operation, emit_terminal);
Self::settle_operation(
&in_flight,
&address,
&operation,
emit_terminal,
Some(permit),
);
});
let Ok(task) = task else {
Self::settle_operation(
&spawned_in_flight,
&spawned_address,
&spawned_operation,
false,
None,
);
return Err(HookOrchestrationAdmissionV1::Unavailable);
};
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -257,6 +257,54 @@ async fn explicit_request_preserves_a_newer_native_event() {
assert!(runtime.shutdown().await);
}

/// A cycle's terminal is the host's cue to send the session's next boundary,
/// so the cycle must release its permit before it reports that terminal.
#[tokio::test]
async fn settled_cycle_admits_the_sessions_next_boundary_from_its_terminal() {
let mut first = native_request(221);
let mut envelope = first.hook.envelope().clone();
envelope.event_id = [229; 16];
let binding = HookScopeBindingV1 {
host: envelope.producer,
project_id: envelope.project_id,
repository_id: envelope.repository_id,
worktree_id: envelope.worktree_id,
worktree_epoch: envelope.worktree_epoch,
binding_token: envelope.binding_token,
capabilities: vec![tracedecay_hooks::HookCapabilityV1 {
family: tracedecay_hooks::HookEventFamily::SavedEdit,
support: tracedecay_hooks::stock_event_support(
envelope.producer,
tracedecay_hooks::HookEventFamily::SavedEdit,
),
}],
};
let next =
HookOrchestrationRequestV1::from_envelope(envelope, &binding, None, 1, false).unwrap();
let runtime = BoundedHookOrchestratorV1::new(1, |_, _| async {
HookOrchestrationWorkOutcomeV1::Completed
})
.unwrap();
let (admitted, admission) = tokio::sync::oneshot::channel();
let pending = StdMutex::new(Some((next, admitted)));
let owner = Arc::downgrade(&runtime);
first.completion = Some(Arc::new(move || {
if let Some((next, admitted)) = pending.lock().unwrap().take()
&& let Some(runtime) = owner.upgrade()
{
let _ = admitted.send(runtime.admit(next));
}
}));

assert_eq!(runtime.admit(first), HookOrchestrationAdmissionV1::Enqueued);
assert_eq!(
admission.await.unwrap(),
HookOrchestrationAdmissionV1::Enqueued,
"a settled cycle must not backpressure the boundary its terminal invites"
);
assert!(runtime.shutdown().await);
}

/// A boundary of a session with no Context Scout lifecycle runs once and
/// settles. It reruns only when a later hook of that session carries a
/// lifecycle, and a newer boundary of the session replaces it.
Expand Down Expand Up @@ -354,6 +402,15 @@ async fn lifecycle_less_boundary_settles_and_resumes_on_its_sessions_lifecycle()
Some(([210; 16], true)),
"only the session's hook that carries a lifecycle resumes the parked boundary"
);
// The resumed cycle reports no terminal. Until it settles, the session's
// next boundary supersedes it and inherits its lifecycle.
tokio::time::timeout(std::time::Duration::from_secs(5), async {
while runtime.active_tasks() != 0 {
tokio::task::yield_now().await;
}
})
.await
.expect("the resumed cycle settles");
session_hook(213, Some(lifecycle.clone()));
let second = saved_edit(214);
assert_eq!(
Expand Down
Loading