From b0733e53b22b51f5e2c6f730393553fa0e7e1b32 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 28 Oct 2026 12:15:35 +0900 Subject: [PATCH] fix: release removed Subject bodies while preserving journal history --- crates/subjektiv/src/journal/binding.rs | 23 + crates/subjektiv/src/journal/store.rs | 26 + crates/workspace-server/src/retention.rs | 24 +- crates/workspace-server/src/server.rs | 71 +- .../src/server/subjektiv_journal.rs | 419 +++++--- .../server/tests/subjektiv_journal_tests.rs | 974 +++++++++++++++++- crates/workspace-server/src/store.rs | 6 +- 7 files changed, 1370 insertions(+), 173 deletions(-) diff --git a/crates/subjektiv/src/journal/binding.rs b/crates/subjektiv/src/journal/binding.rs index 3430a72cb..9267cd50c 100644 --- a/crates/subjektiv/src/journal/binding.rs +++ b/crates/subjektiv/src/journal/binding.rs @@ -194,6 +194,29 @@ impl SubjektivStore { }) } + /// End a body after the Host has confirmed its Runtime removal. This is not a + /// completed Journal handoff: source coverage, positions and saved receipts stay + /// unchanged. Revocation and closing generations are atomic, so no retained Job + /// can write after deletion even while Backend Job cleanup is being retried. + pub fn release_removed_journal_body(&self, host: &JournalHostContext) -> JournalResult<()> { + self.database.try_transaction(|tx| { + super::store::check_session_host(self, tx, host)?; + let attr = host.attribution.as_ref().expect("checked Session authority"); + let binding: Option = tx.query_row( + "SELECT binding_id FROM journal_writer_bindings WHERE subject_id=?1 AND runtime_id=?2 AND worker_id=?3 ORDER BY rowid DESC LIMIT 1", + params![host.subject_id, attr.runtime_id, attr.worker_id], |r| r.get(0), + ).optional()?; + let binding = binding.unwrap_or_else(|| crate::issued_id("journal-removed-body")); + tx.execute("INSERT OR IGNORE INTO journal_writer_bindings(binding_id,subject_id,runtime_id,worker_id,active) VALUES(?1,?2,?3,?4,0)",params![binding,host.subject_id,attr.runtime_id,attr.worker_id])?; + tx.execute("UPDATE journal_writer_bindings SET active=0 WHERE subject_id=?1 AND runtime_id=?2 AND worker_id=?3",params![host.subject_id,attr.runtime_id,attr.worker_id])?; + // Bind legacy attributed Sessions to an inactive lifetime, never to the + // successor. Retained unfinished history is not an active writer. + tx.execute("INSERT OR IGNORE INTO journal_session_bindings(session_id,binding_id) SELECT session_id,?4 FROM subject_session_attributions WHERE subject_id=?1 AND runtime_id=?2 AND worker_id=?3",params![host.subject_id,attr.runtime_id,attr.worker_id,binding])?; + tx.execute("UPDATE journal_generations SET active=0 WHERE journal_id IN (SELECT j.id FROM journals j JOIN subject_session_attributions a ON a.session_id=j.source_session_id WHERE a.subject_id=?1 AND a.runtime_id=?2 AND a.worker_id=?3)",params![host.subject_id,attr.runtime_id,attr.worker_id])?; + Ok(()) + }) + } + /// The Host must hold its lifecycle fence, confirm Idle, reconcile all committed original /// Sessions (including the save/notification gap), and exclude queued/running/Unknown Jobs. /// This transaction rejects retained writers, pending source bytes and active generations. diff --git a/crates/subjektiv/src/journal/store.rs b/crates/subjektiv/src/journal/store.rs index 72008f1b1..a5e4bedf4 100644 --- a/crates/subjektiv/src/journal/store.rs +++ b/crates/subjektiv/src/journal/store.rs @@ -651,6 +651,32 @@ impl SubjektivStore { Ok(()) }) } + /// Read the exact retained Job/attempt snapshot without granting writer + /// authority or creating a generation. Used for outcome recovery after removal. + pub fn retained_journal_generation( + &self, + host: &JournalHostContext, + id: &str, + binding: &JobAttemptBinding, + ) -> JournalResult> { + self.database.try_with_connection(|tx| { + check_host(self, tx, host, false)?; + check_session_host(self, tx, host)?; + journal(tx, host, id, true)?; + let raw: Option = tx.query_row( + "SELECT snapshot_json FROM journal_generations WHERE journal_id=?1 AND job_id=?2 AND attempt_id=?3", + params![id, binding.job_id, binding.attempt_id], |r| r.get(0), + ).optional()?; + raw.map(|raw| { + let generation: JournalGeneration = serde_json::from_str(&raw)?; + if generation.journal_id != id || &generation.job != binding { + return Err(conflict("retained generation identity mismatch")); + } + Ok(generation) + }).transpose() + }) + } + pub fn read_journal_generation( &self, host: &JournalHostContext, diff --git a/crates/workspace-server/src/retention.rs b/crates/workspace-server/src/retention.rs index b8469122e..3b01d553e 100644 --- a/crates/workspace-server/src/retention.rs +++ b/crates/workspace-server/src/retention.rs @@ -325,7 +325,7 @@ impl SqliteWorkspaceStore { let now = Utc::now().to_rfc3339(); self.with_conn_mut(|conn| { let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?; - let policy=load_removal_policy(&tx,&req.workspace_id,&req.worker,erased_journal)?.ok_or_else(|| StoreError::InvalidInput(format!("policy-missing:{}",req.workspace_id)))?; + let policy=load_removal_policy(&tx,&req.workspace_id,&req.worker,erased_journal,true)?.ok_or_else(|| StoreError::InvalidInput(format!("policy-missing:{}",req.workspace_id)))?; let worker=match load_worker(&tx,&req.workspace_id,&req.worker)? { Some(v)=>v, None=>{ @@ -371,7 +371,7 @@ impl SqliteWorkspaceStore { return Err(StoreError::InvalidInput(format!("stale:{plan_id}:plan state {} is not executable", state_s(plan.state)))); } if !plan.blockers.is_empty(){return Err(StoreError::InvalidInput(format!("blocked:{}",serde_json::to_string(&plan.blockers).unwrap())));} - let policy=load_removal_policy(&tx,workspace_id,&plan.worker,plan.policy_id==ERASED_JOURNAL_POLICY_ID)?.ok_or_else(||StoreError::InvalidInput(format!("policy-missing:{workspace_id}")))?; + let policy=load_removal_policy(&tx,workspace_id,&plan.worker,plan.policy_id==ERASED_JOURNAL_POLICY_ID,plan.policy_id=="backend-subject-removal-archive-v1")?.ok_or_else(||StoreError::InvalidInput(format!("policy-missing:{workspace_id}")))?; if policy.policy_id!=plan.policy_id||policy.policy_digest!=plan.policy_digest{ mark_stale(&tx,&plan,"policy content changed")?;tx.commit()?; return Err(stale_error(&plan,"policy content changed")); @@ -1020,6 +1020,7 @@ fn load_removal_policy( w: &str, worker: &RuntimeWorkerRef, erased_journal: bool, + preserve_subject_history: bool, ) -> crate::Result> { let Some(mut policy) = load_policy(c, w)? else { return Ok(None); @@ -1044,6 +1045,25 @@ fn load_removal_policy( policy.archive_retention = update.archive_retention; policy.diagnostics_disposition = update.diagnostics_disposition; policy.diagnostics_retention_seconds = None; + } else if preserve_subject_history && policy.session_disposition == SessionDisposition::Purge && c.query_row( + "SELECT EXISTS(SELECT 1 FROM worker_create_reservations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND substr(singleton_key,1,10)='subjektiv:')", + params![w, worker.runtime_id, worker.worker_id], + |r| r.get::<_, bool>(0), + )? { + // Removing a Subject body releases occupancy, not its original history. + // A configured Purge must not silently erase an unfinished Journal. Explicit + // Session erasure uses the separate, proof-bound branch above. + let update = WorkerRetentionPolicyUpdate { + policy_id: "backend-subject-removal-archive-v1".into(), + session_disposition: SessionDisposition::Archive, + metadata_disposition: policy.metadata_disposition, + archive_retention: policy.archive_retention, + diagnostics_disposition: policy.diagnostics_disposition, + diagnostics_retention_seconds: policy.diagnostics_retention_seconds, + }; + policy.policy_id = update.policy_id.clone(); + policy.policy_digest = worker_retention_policy_digest(&update); + policy.session_disposition = update.session_disposition; } Ok(Some(policy)) } diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index cc505f91a..bca0a9580 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -2327,16 +2327,29 @@ impl WorkerRemovalService { )); } }; - match self.worker_projection.publish_ordered(|| { - self.store - .commit_worker_removal( - &self.workspace_id, - &prepared.plan.operation_id, - &prepared.plan.input_fingerprint, - &result, - ) - .map(|commit| ((), commit.catalog)) - }) { + let commit_result = + match subjektiv_journal::finish_worker_removal(self, target, &prepared, &result, || { + self.worker_projection.publish_ordered(|| { + self.store + .commit_worker_removal( + &self.workspace_id, + &prepared.plan.operation_id, + &prepared.plan.input_fingerprint, + &result, + ) + .map(|commit| ((), commit.catalog)) + }) + }) { + Ok(result) => result, + Err(error) => { + return Ok(worker_remove_error_response( + StatusCode::CONFLICT, + "subject_removal_completion_unconfirmed", + &error.to_string(), + )); + } + }; + match commit_result { Ok(()) => { subjektiv_journal::after_worker_removal(self, &result); Ok(worker_remove_success_response(target)) @@ -2722,16 +2735,34 @@ impl WorkerRemovalService { )); } }; - match self.worker_projection.publish_ordered(|| { - self.store - .commit_worker_removal( - &self.workspace_id, - &plan.operation_id, - &plan.input_fingerprint, - &retention_result, - ) - .map(|commit| ((), commit.catalog)) - }) { + let commit_result = match subjektiv_journal::finish_worker_removal( + self, + target, + &prepared, + &retention_result, + || { + self.worker_projection.publish_ordered(|| { + self.store + .commit_worker_removal( + &self.workspace_id, + &plan.operation_id, + &plan.input_fingerprint, + &retention_result, + ) + .map(|commit| ((), commit.catalog)) + }) + }, + ) { + Ok(result) => result, + Err(error) => { + return Ok(worker_remove_error_response( + StatusCode::CONFLICT, + "subject_removal_completion_unconfirmed", + &error.to_string(), + )); + } + }; + match commit_result { Ok(()) => {} Err(error) => { let _ = self.store.fail_worker_removal( diff --git a/crates/workspace-server/src/server/subjektiv_journal.rs b/crates/workspace-server/src/server/subjektiv_journal.rs index fccee17a6..fded1c5fb 100644 --- a/crates/workspace-server/src/server/subjektiv_journal.rs +++ b/crates/workspace-server/src/server/subjektiv_journal.rs @@ -92,6 +92,20 @@ pub(super) fn prepare(api: &WorkspaceApi, reservation: &BackendJobReservation) - }; let store = store(api)?; let host = host(api, &store, grant)?; + let attr = store + .session_attribution(&grant.session_id) + .map_err(|e| Error::Store(e.to_string()))? + .ok_or_else(|| { + Error::WorkspacePermissionDenied("Journal Session attribution missing".into()) + })?; + if store + .journal_worker_was_released(&attr.subject_id, &attr.runtime_id, &attr.worker_id) + .map_err(domain)? + { + return Err(Error::WorkspacePermissionDenied( + "released Journal body cannot dispatch another attempt".into(), + )); + } let input = saved_input(&reservation.job.request, grant)?; let g = store .prepare_journal_input( @@ -584,6 +598,18 @@ fn dispatch( .lock() .map_err(|_| Error::Store("Journal lifecycle lock poisoned".into()))?; let store = store(api)?; + if store + .journal_worker_was_released( + &attribution.subject_id, + &attribution.runtime_id, + &attribution.worker_id, + ) + .map_err(domain)? + { + // Deletion can retain an unfinished prefix. It must not mint attempts or + // transfer the removed body's write capability to its successor. + return Ok(()); + } let host = JournalHostContext::validated_session(&store, attribution.clone()).map_err(domain)?; if let Some(g) = store @@ -1740,8 +1766,8 @@ fn finish_erased_release( .release_subject_singleton(api.workspace_id(), &lease, binding_id, &operation) } -/// Normal deletion cannot substitute for Subject release, even for a stopped or disabled body. -/// Explicit erasure remains a separate monotonic source/diagnostics destruction path. +/// Ordinary removal owns writer revocation and occupancy release. Before execution +/// only preserve/erase source requirements are checked; an active writer is not a blocker. pub(super) fn prepare_worker_removal( service: &WorkerRemovalService, runtime: &RuntimeRegistry, @@ -1762,7 +1788,6 @@ pub(super) fn prepare_worker_removal( &service.journal_registration, ) .map_err(|e| Error::Store(e.to_string()))?; - let host = JournalHostContext::validated_session(&store, attr.clone()).map_err(domain)?; let source = runtime_api::SessionPublicSource::Retained { worker_id: worker_runtime::identity::WorkerId::parse(&target.worker_id) .ok_or_else(|| Error::Store("Worker identity invalid".into()))?, @@ -1779,25 +1804,9 @@ pub(super) fn prepare_worker_removal( finish_erased_release(&api, &attr)?; return Ok(Some(proof)); } - let Some(binding_id) = host.writer_binding_id() else { - return Err(Error::RepositoryConflict( - "Subject writer evidence is missing; ordinary deletion cannot bypass release".into(), - )); - }; - let writer = JournalWriterBinding { - subject_id: attr.subject_id.clone(), - runtime_id: attr.runtime_id.clone(), - worker_id: attr.worker_id.clone(), - binding_id: binding_id.into(), - }; - if store.journal_writer_active(&writer).map_err(domain)? { - return Err(Error::RepositoryConflict("Subject remains occupied: restore to actual Idle, finish Journal recording, then explicitly release before ordinary deletion".into())); - } - if store.journal_retention_required(&host).map_err(domain)? { - return Err(Error::RepositoryConflict( - "Released Subject has unresolved Journal material; retain occupancy and source".into(), - )); - } + // Do not require a live Controller, complete Journal, or prior release here. + // Runtime removal is the physical fence; finish_worker_removal revokes the + // exact old body only after its successful, matching retention receipt. Ok(None) } @@ -1886,20 +1895,6 @@ pub(super) fn validate_removal_retention( // its retained permanent fence, including recovery after the aggregate is gone. return Ok(()); } - let binding_id = host.writer_binding_id().ok_or_else(|| { - Error::RepositoryConflict("Subject writer evidence missing during removal recovery".into()) - })?; - let binding = JournalWriterBinding { - subject_id: attr.subject_id.clone(), - runtime_id: attr.runtime_id.clone(), - worker_id: attr.worker_id.clone(), - binding_id: binding_id.into(), - }; - if store.journal_writer_active(&binding).map_err(domain)? { - return Err(Error::RepositoryConflict( - "Prepared removal cannot bypass explicit Subject release".into(), - )); - } let source = runtime_api::SessionPublicSource::Retained { worker_id: worker_runtime::identity::WorkerId::parse(&target.worker_id) .ok_or_else(|| Error::Store("Worker identity invalid".into()))?, @@ -1907,37 +1902,140 @@ pub(super) fn validate_removal_retention( let runtime = service.runtime.upgrade().ok_or_else(|| { Error::Store("Runtime unavailable for source retention confirmation".into()) })?; - if store.journal_retention_required(&host).map_err(domain)? { - if prepared.runtime_request.session_disposition - != worker_runtime::retention::SessionDisposition::Archive - || prepared.runtime_request.archive_id.is_none() - { - return Err(Error::RepositoryConflict( - "unreflected Journal requires a retained original Session archive".into(), - )); + let retention_required = store.journal_retention_required(&host).map_err(domain)?; + if prepared.runtime_request.session_disposition + == worker_runtime::retention::SessionDisposition::Purge + && prepared.runtime_request.archive_id.is_none() + && !retention_required + && store + .journal_worker_was_released(&attr.subject_id, &attr.runtime_id, &attr.worker_id) + .map_err(domain)? + { + // Compatibility for an already issued, history-complete ordinary Purge + // plan. New Subject plans always Archive. Keep the old fingerprint and + // disposition; never turn this into an erasure or allow an active writer. + // A missing Runtime Worker is NOT success: execute_worker_retention still + // has to return the exact persisted receipt before anything can commit. + // A pending receipt may outlive the physical aggregate while Runtime's + // in-memory catalog still contains the Worker. Pin release is best effort + // in this historical branch; exact Runtime retention replay still enforces + // pins if the source exists, and requires a receipt if it is already gone. + if let Err(error) = retain_source(&runtime, &service.workspace_id, &attr, source, true) { + tracing::debug!(%error, "Historical Purge source unavailable; reconcile exact Runtime receipt"); } - if let Err(original) = retain_source(&runtime, &service.workspace_id, &attr, source, false) - { - // Recovery may already have transferred the physical source before its Backend - // catalog commit; authenticate the exact reserved archive, never another Session. - let archive = runtime_api::SessionPublicSource::Archived { - archive_id: prepared - .runtime_request - .archive_id - .clone() - .expect("checked"), - }; - retain_source(&runtime, &service.workspace_id, &attr, archive, false) - .map_err(|_| original)?; - } - } else { - // This stopped/removal-fenced Session has no admitted remainder; purge need not - // force a generation. A failed release conservatively blocks ordinary deletion. - retain_source(&runtime, &service.workspace_id, &attr, source, true)?; + return Ok(()); + } + if prepared.runtime_request.session_disposition + != worker_runtime::retention::SessionDisposition::Archive + || prepared.runtime_request.archive_id.is_none() + { + return Err(Error::RepositoryConflict( + "Subject removal retains its original Session; explicit erasure is required for Purge" + .into(), + )); + } + let release = !retention_required; + if let Err(original) = retain_source(&runtime, &service.workspace_id, &attr, source, release) { + // A lost reply may follow the physical transfer. Reconcile only the exact + // reserved archive and Session, whether or not an admitted remainder exists. + let archive = runtime_api::SessionPublicSource::Archived { + archive_id: prepared + .runtime_request + .archive_id + .clone() + .expect("checked"), + }; + retain_source(&runtime, &service.workspace_id, &attr, archive, release) + .map_err(|_| original)?; } Ok(()) } +/// A successful Runtime receipt, not catalog liveness, authorizes ending the old +/// writer. The ordinary removal commit releases occupancy atomically with catalog +/// deletion. Until then a crash leaves the singleton occupied and the same receipt +/// can safely repeat this step; no Controller restore or Journal generation is needed. +pub(super) fn finish_worker_removal( + service: &WorkerRemovalService, + target: &RuntimeWorkerRef, + prepared: &crate::retention::PreparedWorkerRemoval, + result: &worker_runtime::retention::WorkerRetentionExecutionResult, + commit: impl FnOnce() -> T, +) -> Result { + let Some(attr) = removal_attribution(service, target)? else { + return Ok(commit()); + }; + let request = &prepared.runtime_request; + if prepared.plan.worker != *target + || request.workspace_id != service.workspace_id + || request.source_runtime_id != target.runtime_id + || request.worker_id.to_string() != target.worker_id + || !result.source_removed + || result.worker_id != request.worker_id + || result.operation_id != request.operation_id + || result.input_fingerprint != request.input_fingerprint + || result.session_disposition != request.session_disposition + || result.diagnostics_disposition != request.diagnostics_disposition + || result.diagnostics_retained + != (request.diagnostics_disposition + == worker_runtime::retention::DiagnosticsDisposition::Retain) + { + return Err(Error::RepositoryConflict( + "Subject removal receipt does not match its plan".into(), + )); + } + if request.session_disposition == worker_runtime::retention::SessionDisposition::Archive { + let archive = result.archive.as_ref().ok_or_else(|| { + Error::RepositoryConflict("Subject removal archive receipt is missing".into()) + })?; + if Some(&archive.archive_id) != request.archive_id.as_ref() + || archive.workspace_id != request.workspace_id + || archive.source_runtime_id != request.source_runtime_id + || archive.source_worker_id != request.worker_id + || archive.source_session_id != attr.session_id + || archive.operation_id != request.operation_id + || archive.input_fingerprint != request.input_fingerprint + || archive.policy_id != request.policy_id + { + return Err(Error::RepositoryConflict( + "Subject removal archive identity mismatch".into(), + )); + } + } + let api = service.journal_dispatch_api.upgrade().ok_or_else(|| { + Error::Store("Workspace API unavailable for Subject removal completion".into()) + })?; + let gate = lifecycle_lock(&api, &attr.subject_id)?; + let _guard = gate + .lock() + .map_err(|_| Error::Store("Journal lifecycle lock poisoned".into()))?; + let store = store(&api)?; + let host = JournalHostContext::validated_session(&store, attr.clone()).map_err(domain)?; + if store.journal_session_erased(&host).map_err(domain)? { + if request.session_disposition == worker_runtime::retention::SessionDisposition::Archive { + // Erasure may win after Runtime archive transfer. Do not publish that + // archive or skip writer revocation: the same removal retry must take + // erased_archive_removal_proof and its physical-erasure commit path. + return Err(Error::RepositoryConflict( + "Session erased during removal; reconcile the same removal operation".into(), + )); + } + return Ok(commit()); + } + store.release_removed_journal_body(&host).map_err(domain)?; + for session in store + .subject_session_attributions(&attr.subject_id) + .map_err(|e| Error::Store(e.to_string()))? + .into_iter() + .filter(|a| a.runtime_id == target.runtime_id && a.worker_id == target.worker_id) + { + reconcile_session_jobs(&api, &session, false)?; + } + // Keep erasure excluded through the catalog commit: otherwise a tombstone + // could win between revocation and publication of the retained archive. + Ok(commit()) +} + #[derive(Deserialize)] #[serde(deny_unknown_fields)] pub(super) struct Empty {} @@ -2089,7 +2187,28 @@ pub(super) fn reconcile_unknown( .as_ref() .ok_or_else(|| Error::InvalidInput("Journal recovery requires a grant".into()))?; let store = store(api)?; - let host = host(api, &store, grant)?; + let attr = store + .session_attribution(&grant.session_id) + .map_err(|e| Error::Store(e.to_string()))? + .filter(|a| a.subject_id == grant.subject_id) + .ok_or_else(|| Error::WorkspacePermissionDenied("Journal attribution mismatch".into()))?; + let released = store + .journal_worker_was_released(&attr.subject_id, &attr.runtime_id, &attr.worker_id) + .map_err(domain)?; + let host = if released { + // Recovery reads immutable evidence, never renews a legacy writer grant. + let host = JournalHostContext::validated_session(&store, attr.clone()).map_err(domain)?; + if !grant.binding_id.is_empty() + && host.writer_binding_id() != Some(grant.binding_id.as_str()) + { + return Err(Error::WorkspacePermissionDenied( + "Journal recovery binding mismatch".into(), + )); + } + host + } else { + host(api, &store, grant)? + }; let erased = store.journal_session_erased(&host).map_err(domain)?; if !erased { job.load_executable_request()?; @@ -2097,70 +2216,93 @@ pub(super) fn reconcile_unknown( job_id: job.request.job_id.clone(), attempt_id: attempt.attempt_id.clone(), }; - let g = if legacy_descriptor(&job.request) { - // Only reconcile already-saved legacy receipts; never redispatch descriptors. + let generation = if released { store - .prepare_journal_generation( - &host, - &grant.journal_id, - binding, - grant.selected_ranges.clone(), - ) + .retained_journal_generation(&host, &grant.journal_id, &binding) .map_err(domain)? + } else if legacy_descriptor(&job.request) { + Some( + store + .prepare_journal_generation( + &host, + &grant.journal_id, + binding, + grant.selected_ranges.clone(), + ) + .map_err(domain)?, + ) } else { - let input = saved_input(&job.request, grant)?; - let generation = store - .prepare_journal_input(&host, &input, binding) - .map_err(domain)?; - input.validate_generation(&generation).map_err(domain)?; - generation + // Preserve active-body recovery: its saved intent can still establish + // the original generation fence. A released body must only read it. + Some( + store + .prepare_journal_input(&host, &saved_input(&job.request, grant)?, binding) + .map_err(domain)?, + ) }; - let ctx = JournalJobContext::validated_recovery(&store, &host, &g.id, g.job.clone()) - .map_err(domain)?; - match journal::service::finish(&store, &ctx) { - Ok(completion) => { - let submission = BackendJobResultSubmission { - job_id: job.request.job_id.clone(), - attempt_id: attempt.attempt_id.clone(), - input_digest: attempt.input_digest.clone(), - result: serde_json::to_value(completion) - .map_err(|e| Error::Store(e.to_string()))?, - }; - let current = api.store.reconcile_backend_journal_attempt( - api.workspace_id(), - attempt, - Some(&submission), - &Utc::now().to_rfc3339(), - )?; - let saved = api - .store - .get_backend_job(api.workspace_id(), &job.request.job_id)? - .ok_or_else(|| Error::Store("reconciled Job missing".into()))?; - notification_delivery::deliver_job_result( - api, - &BackendJobResultAcceptance { - job: saved, - attempt: current.clone(), - replayed: false, - }, - ); - if current.worker_cleanup_state == Some(BackendJobWorkerCleanupState::Completed) { - drain_after_completion(api, &job.request); - if job.input_disposition == crate::backend_job::JobInputDisposition::Available { - api.store.reclaim_backend_job_input( - api.workspace_id(), - &job.request.job_id, - &job.input_digest, - crate::backend_job::JobInputRetirementReason::Completed, - )?; + if let Some(g) = generation { + if g.selected_ranges != grant.selected_ranges { + return Err(Error::WorkspacePermissionDenied( + "Journal recovery selection mismatch".into(), + )); + } + if !legacy_descriptor(&job.request) { + saved_input(&job.request, grant)? + .validate_generation(&g) + .map_err(domain)?; + } + let ctx = JournalJobContext::validated_recovery(&store, &host, &g.id, g.job.clone()) + .map_err(domain)?; + match journal::service::finish(&store, &ctx) { + Ok(completion) => { + let submission = BackendJobResultSubmission { + job_id: job.request.job_id.clone(), + attempt_id: attempt.attempt_id.clone(), + input_digest: attempt.input_digest.clone(), + result: serde_json::to_value(completion) + .map_err(|e| Error::Store(e.to_string()))?, + }; + let current = api.store.reconcile_backend_journal_attempt( + api.workspace_id(), + attempt, + Some(&submission), + &Utc::now().to_rfc3339(), + )?; + let saved = api + .store + .get_backend_job(api.workspace_id(), &job.request.job_id)? + .ok_or_else(|| Error::Store("reconciled Job missing".into()))?; + notification_delivery::deliver_job_result( + api, + &BackendJobResultAcceptance { + job: saved, + attempt: current.clone(), + replayed: false, + }, + ); + if current.worker_cleanup_state == Some(BackendJobWorkerCleanupState::Completed) + { + if !released { + drain_after_completion(api, &job.request); + } + if job.input_disposition + == crate::backend_job::JobInputDisposition::Available + { + api.store.reclaim_backend_job_input( + api.workspace_id(), + &job.request.job_id, + &job.input_digest, + crate::backend_job::JobInputRetirementReason::Completed, + )?; + } + } else { + api.schedule_backend_job_worker_cleanup(¤t); } - } else { - api.schedule_backend_job_worker_cleanup(¤t); + return Ok(()); } - return Ok(()); + Err(JournalError::Conflict(_) | JournalError::PermissionDenied(_)) => {} + Err(error) => return Err(domain(error)), } - Err(JournalError::Conflict(_) | JournalError::PermissionDenied(_)) => {} - Err(error) => return Err(domain(error)), } } if attempt.worker_cleanup_state != Some(BackendJobWorkerCleanupState::Completed) { @@ -2179,11 +2321,17 @@ pub(super) fn reconcile_unknown( false }; if !confirmed_purge { + if released { + // Source deletion already fenced this generation. An unbound Unknown + // still lacks execution-cleanup proof: preserve Unknown, but it no + // longer owns the Subject or blocks removal of the original body. + return Ok(()); + } return Err(Error::RepositoryConflict( "Unknown Journal attempt has no committed stop/Purge fence; not retried".into(), )); } - if !erased { + if !erased && !released { close(api, &job.request, attempt)?; } api.store.reconcile_backend_journal_attempt( @@ -2192,7 +2340,7 @@ pub(super) fn reconcile_unknown( None, &Utc::now().to_rfc3339(), )?; - if !erased { + if !erased && !released { // Resume the same saved body input only after canonical execution/domain fences. // dispatch selects its retained intent before it can capture any newer source. let attr = store @@ -2209,6 +2357,10 @@ fn reconcile_session_jobs( attr: &crate::subjektiv::SubjectSessionAttribution, erased: bool, ) -> Result<()> { + let released = !erased + && store(api)? + .journal_worker_was_released(&attr.subject_id, &attr.runtime_id, &attr.worker_id) + .map_err(domain)?; let mut after: Option = None; loop { let jobs = api.store.list_backend_journal_jobs( @@ -2223,6 +2375,29 @@ fn reconcile_session_jobs( .store .get_backend_job_attempt(api.workspace_id(), &job.request.job_id, &id)? .ok_or_else(|| Error::Store("Journal current attempt missing".into()))?; + if released { + // Writer revocation has already fenced all Journal mutations. End + // pending/Unknown attempts without claiming their input was reflected; + // saved input, operations and original source remain retained. + if !attempt.state.terminal() { + attempt = api.store.finish_backend_job_attempt( + api.workspace_id(), + &job.request.job_id, + &id, + BackendJobAttemptState::Failed, + "journal_body_released", + "Source Worker was removed; unfinished Journal history retained", + &Utc::now().to_rfc3339(), + )?; + } + api.schedule_backend_job_worker_cleanup(&attempt); + if attempt.state == BackendJobAttemptState::Unknown { + // A saved checkpoint may settle success; otherwise retain Unknown + // until canonical Job-worker cleanup has supplied its Purge fence. + reconcile_unknown(api, job, &attempt)?; + } + continue; + } if !erased && legacy_descriptor(&job.request) && job.input_disposition == crate::backend_job::JobInputDisposition::Available diff --git a/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs b/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs index 364438db2..2d8ef5414 100644 --- a/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs +++ b/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs @@ -1134,6 +1134,7 @@ async fn unknown_partial_waits_for_durable_purge_fence_and_retains_sources_when_ "a stale cleanup snapshot must not overwrite newer evidence" ); subjektiv_journal::reconcile_unknown(&f.api, &dispatched.job, &unknown).unwrap(); + f.execution.take_inputs(); f.api .store .update_worker_retention( @@ -1151,8 +1152,9 @@ async fn unknown_partial_waits_for_durable_purge_fence_and_retains_sources_when_ BackendJobWorkerCleanupState::Completed, ) .await; - // Cleanup commits before domain reconciliation; wait for its authoritative outcome. - tokio::time::timeout(std::time::Duration::from_secs(10), async { + // Cleanup automatically retries the same immutable input after reconciliation. + // Failed is transient: a second manual retry races the production continuation. + let retry = tokio::time::timeout(std::time::Duration::from_secs(10), async { loop { let job = f .api @@ -1160,8 +1162,22 @@ async fn unknown_partial_waits_for_durable_purge_fence_and_retains_sources_when_ .get_backend_job(TEST_WORKSPACE_ID, &f.request.job_id) .unwrap() .unwrap(); - if job.state == BackendJobState::Failed { - break; + if job.current_attempt == 2 { + let id = crate::backend_job::attempt_id(&f.request.job_id, 2); + let attempt = f + .api + .store + .get_backend_job_attempt(TEST_WORKSPACE_ID, &f.request.job_id, &id) + .unwrap() + .unwrap(); + if attempt.state == BackendJobAttemptState::Dispatched { + break BackendJobReservation { + job, + attempt, + replayed: true, + resource_reused: false, + }; + } } tokio::task::yield_now().await; } @@ -1187,19 +1203,8 @@ async fn unknown_partial_waits_for_durable_purge_fence_and_retains_sources_when_ 1, "source pin survives failed/partial Job cleanup" ); - f.execution.take_inputs(); - let retry = f - .api - .store - .reserve_backend_job_retry( - TEST_WORKSPACE_ID, - &f.request.job_id, - &Utc::now().to_rfc3339(), - ) - .unwrap(); assert_eq!(retry.job.request.input, f.request.input); assert_eq!(retry.job.input_digest, f.request.input_digest().unwrap()); - let retry = f.api.dispatch_backend_job_reservation(retry).unwrap(); let resumed = retry.attempt.worker.as_ref().unwrap(); let delivered = f.execution.take_inputs(); assert_eq!(delivered.len(), 1); @@ -2018,6 +2023,114 @@ impl SubjectRemovalFixture { .release_subject_singleton(TEST_WORKSPACE_ID, &lease, &binding.binding_id, operation) .unwrap(); } + fn pending_journal(&self) -> BackendJobReservation { + self.pending_journal_input(false) + } + + fn pending_journal_input(&self, legacy: bool) -> BackendJobReservation { + let capture = + subjektiv_journal::capture_input(&self.api, &self.attr, 0, 0, 48 * 1024).unwrap(); + let ranges = subjektiv::journal::recording::capture_ranges(&capture).unwrap(); + let id = self + .store + .accept_journal_sources(&self.host, "unfinished history", &ranges) + .unwrap(); + let input = subjektiv::journal::recording::JournalInput::build( + &self.store, + &self.host, + &id, + |order, offset, budget| { + subjektiv_journal::capture_input(&self.api, &self.attr, order, offset, budget) + .map_err(|e| JournalError::Conflict(e.to_string())) + }, + ) + .unwrap(); + let mut request = + subjektiv_journal::recording_request(&self.host, self.attr.clone(), &id, input) + .unwrap(); + if legacy { + let grant = request.grants.subjektiv_journal.as_mut().unwrap(); + grant.binding_id.clear(); + request.input = + serde_json::json!({"journal_id":id, "selected_ranges":grant.selected_ranges}); + } + let reserved = self + .api + .store + .reserve_backend_job(TEST_WORKSPACE_ID, &request, &Utc::now().to_rfc3339()) + .unwrap(); + if legacy { + self.store + .prepare_journal_generation( + &self.host, + &id, + subjektiv::api::JobAttemptBinding { + job_id: request.job_id.clone(), + attempt_id: reserved.attempt.attempt_id.clone(), + }, + request + .grants + .subjektiv_journal + .as_ref() + .unwrap() + .selected_ranges + .clone(), + ) + .unwrap(); + } else { + subjektiv_journal::prepare(&self.api, &reserved).unwrap(); + } + reserved + } + + fn assert_removed_with_history(&self) -> String { + assert!( + self.api + .store + .get_worker_registry(TEST_WORKSPACE_ID, &self.target) + .unwrap() + .is_none() + ); + let key = crate::subjektiv::subject_worker_singleton_key(&self.attr.subject_id).unwrap(); + assert!( + self.api + .store + .current_worker_singleton_owner(TEST_WORKSPACE_ID, &key) + .unwrap() + .is_none() + ); + assert!( + self.store + .journal_worker_was_released( + &self.attr.subject_id, + &self.target.runtime_id, + &self.target.worker_id + ) + .unwrap() + ); + assert!(!self.store.journal_session_erased(&self.host).unwrap()); + assert!(!session_root(self.dir.path(), &self.target).exists()); + self.api + .config_store + .worker_session_archive(TEST_WORKSPACE_ID, &self.target, &self.attr.session_id) + .unwrap() + .expect("ordinary deletion retains original history") + .archive_id + } + + fn assert_occupied(&self) { + let key = crate::subjektiv::subject_worker_singleton_key(&self.attr.subject_id).unwrap(); + assert_eq!( + self.api + .store + .current_worker_singleton_owner(TEST_WORKSPACE_ID, &key) + .unwrap() + .unwrap() + .worker, + self.target + ); + } + fn remove(&self) -> worker::WorkspaceResponse { let key = self .api @@ -2067,6 +2180,811 @@ async fn subject_release_never_substitutes_catalog_idle_or_stopped_for_controlle assert!(f.store.journal_for_session(&f.host).unwrap().is_none()); } +#[tokio::test] +async fn stopped_subject_removal_releases_occupancy_and_preserves_unfinished_history() { + let f = SubjectRemovalFixture::new().await; + let reserved = f.pending_journal(); + let grant = reserved + .job + .request + .grants + .subjektiv_journal + .as_ref() + .unwrap(); + let before = f.store.read_journal(&f.host, &grant.journal_id).unwrap(); + // Even a Workspace Purge policy is not implicit authority to erase Subject history. + let policy = f + .api + .config_store + .worker_retention_policy(TEST_WORKSPACE_ID) + .unwrap() + .unwrap(); + let purge = f + .api + .config_store + .update_worker_retention_policy( + TEST_WORKSPACE_ID, + &policy.policy_digest, + &crate::retention::WorkerRetentionPolicyUpdate { + policy_id: "test-purge".into(), + session_disposition: worker_runtime::retention::SessionDisposition::Purge, + metadata_disposition: policy.metadata_disposition, + archive_retention: policy.archive_retention, + diagnostics_disposition: policy.diagnostics_disposition, + diagnostics_retention_seconds: policy.diagnostics_retention_seconds, + }, + ) + .unwrap(); + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::OK.as_u16(), + "{}", + response.body + ); + let archive = f.assert_removed_with_history(); + let runtime_api::SessionMaterialCatalogAvailability::Page { page } = f + .api + .runtime + .session_material_catalog( + &f.target.runtime_id, + runtime_api::SessionMaterialCatalogRequest { + workspace_id: TEST_WORKSPACE_ID.into(), + source: runtime_api::SessionPublicSource::Archived { + archive_id: archive, + }, + expected_session_id: f.attr.session_id.clone(), + expected_catalog_sha256: None, + offset: 0, + limit: 100, + }, + ) + .unwrap() + else { + panic!("original history must remain readable") + }; + assert_eq!(page.descriptors.len(), 2); + let after = f.store.read_journal(&f.host, &grant.journal_id).unwrap(); + assert_eq!(before.process_order, after.process_order); + assert_eq!(before.process_byte, after.process_byte); + assert!(f.store.journal_retention_required(&f.host).unwrap()); + assert!( + f.store + .active_journal_generation(&f.host, &grant.journal_id) + .is_err() + ); + assert!(subjektiv_journal::prepare(&f.api, &reserved).is_err()); + let job = f + .api + .store + .get_backend_job(TEST_WORKSPACE_ID, &reserved.job.request.job_id) + .unwrap() + .unwrap(); + assert_eq!(job.state, BackendJobState::Failed); + assert_eq!( + job.load_executable_request().unwrap().input, + reserved.job.request.input + ); + assert_eq!( + f.api + .config_store + .worker_retention_policy(TEST_WORKSPACE_ID) + .unwrap() + .unwrap(), + purge + ); + + let Json(next) = scoped_create_workspace_worker( + State(f.api.clone()), + AxumPath(ScopedWorkspacePath { + workspace_id: TEST_WORKSPACE_ID.into(), + }), + HeaderMap::new(), + Json(CreateWorkspaceWorkerRequest { + runtime_id: f.target.runtime_id.clone(), + display_name: "replacement".into(), + singleton_key: None, + profile: Some("builtin:companion".into()), + ticket_assignment: None, + initial_submit: vec![], + workdir_attachments: vec![], + feature_connections: server_api::WorkspaceWorkerFeatureConnectionsRequest { + subjektiv: Some(server_api::SubjektivWorkerConnectionRequest { + subject_id: f.attr.subject_id.clone(), + }), + }, + control_operation_id: None, + execution_grant: None, + }), + ) + .await + .unwrap(); + assert_ne!(next.worker_id, f.target.worker_id); + let key = crate::subjektiv::subject_worker_singleton_key(&f.attr.subject_id).unwrap(); + let successor = f + .api + .store + .current_worker_singleton_owner(TEST_WORKSPACE_ID, &key) + .unwrap() + .unwrap(); + assert_eq!(successor.worker.worker_id, next.worker_id); + let replay = WorkerRemovalService::new(&f.api) + .execute_target_removal_on_host( + f.api.runtime.as_ref(), + &f.target, + "repeat old deletion", + None, + ) + .await + .unwrap(); + assert_eq!(replay.status, StatusCode::OK.as_u16(), "{}", replay.body); + assert_eq!( + f.api + .store + .current_worker_singleton_owner(TEST_WORKSPACE_ID, &key) + .unwrap() + .unwrap(), + successor + ); + assert!(subjektiv_journal::bind_host_session(&f.api, f.attr.clone()).is_err()); + subjektiv_journal::drain_after_completion(&f.api, &reserved.job.request); + assert_eq!( + f.api + .store + .get_backend_job(TEST_WORKSPACE_ID, &reserved.job.request.job_id) + .unwrap() + .unwrap() + .current_attempt, + 1 + ); +} + +#[tokio::test] +async fn subject_removal_archive_failure_does_not_release_writer_or_occupancy() { + let f = SubjectRemovalFixture::new().await; + let archive_parent = f + .dir + .path() + .join(".test-embedded-runtime-store/archives/workers"); + std::fs::create_dir_all(archive_parent.parent().unwrap()).unwrap(); + std::fs::write(&archive_parent, "injected archive failure").unwrap(); + assert_ne!(f.remove().status, StatusCode::OK.as_u16()); + f.assert_occupied(); + assert!( + !f.store + .journal_worker_was_released( + &f.attr.subject_id, + &f.target.runtime_id, + &f.target.worker_id + ) + .unwrap() + ); + assert!(session_root(f.dir.path(), &f.target).exists()); + let old = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + std::fs::remove_file(archive_parent).unwrap(); + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::OK.as_u16(), + "{}", + response.body + ); + f.assert_removed_with_history(); + assert_eq!( + f.api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap() + .plan + .operation_id, + old.plan.operation_id + ); +} + +#[tokio::test] +async fn subject_removal_reopens_after_writer_revocation_before_catalog_commit() { + let f = SubjectRemovalFixture::new().await; + let conn = rusqlite::Connection::open(&f.api.config.database_path).unwrap(); + conn.execute_batch("CREATE TRIGGER interrupt_subject_delete BEFORE DELETE ON worker_registry BEGIN SELECT RAISE(ABORT,'injected metadata failure'); END;").unwrap(); + assert_ne!(f.remove().status, StatusCode::OK.as_u16()); + assert!(!session_root(f.dir.path(), &f.target).exists()); + assert!( + f.store + .journal_worker_was_released( + &f.attr.subject_id, + &f.target.runtime_id, + &f.target.worker_id + ) + .unwrap() + ); + f.assert_occupied(); // Runtime deletion alone is not a successful Backend deletion. + conn.execute_batch("DROP TRIGGER interrupt_subject_delete;") + .unwrap(); + drop(conn); + let f = f.reopen().await; + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::OK.as_u16(), + "{}", + response.body + ); + f.assert_removed_with_history(); +} + +#[tokio::test] +async fn subject_removal_rejects_mismatched_runtime_receipts_before_revocation() { + let f = SubjectRemovalFixture::new().await; + let inventory = f.api.runtime.worker_retention_inventory(&f.target).unwrap(); + let plan = f + .api + .config_store + .plan_worker_removal( + &crate::retention::WorkerRemovalPlanRequest { + workspace_id: TEST_WORKSPACE_ID.into(), + worker: f.target.clone(), + reason: "receipt recovery".into(), + journal_erasure: None, + }, + &inventory, + ) + .unwrap(); + let prepared = f + .api + .config_store + .prepare_worker_removal_execution(TEST_WORKSPACE_ID, &plan.plan_id, &plan.input_fingerprint) + .unwrap(); + let service = WorkerRemovalService::new(&f.api); + subjektiv_journal::validate_removal_retention(&service, &f.target, &prepared).unwrap(); + let result = f + .api + .runtime + .execute_worker_retention(&f.target, prepared.runtime_request.clone()) + .unwrap(); + for variant in 0..5 { + let mut wrong = result.clone(); + match variant { + 0 => wrong.source_removed = false, + 1 => wrong.operation_id.push_str("-wrong"), + 2 => wrong.input_fingerprint.push_str("-wrong"), + 3 => wrong + .archive + .as_mut() + .unwrap() + .source_session_id + .push_str("-wrong"), + _ => wrong.archive = None, + } + assert!( + subjektiv_journal::finish_worker_removal(&service, &f.target, &prepared, &wrong, || ()) + .is_err() + ); + f.assert_occupied(); + assert!( + !f.store + .journal_worker_was_released( + &f.attr.subject_id, + &f.target.runtime_id, + &f.target.worker_id + ) + .unwrap() + ); + } + // Simulate a lost Runtime reply: ordinary retry must recover its exact receipt. + let f = f.reopen().await; + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::OK.as_u16(), + "{}", + response.body + ); + f.assert_removed_with_history(); +} + +#[tokio::test] +async fn subject_removal_unknown_job_settles_after_cleanup_without_restarting_old_writer() { + let f = SubjectRemovalFixture::new().await; + let reserved = f.pending_journal(); + let dispatched = f + .api + .dispatch_backend_job_reservation(reserved.clone()) + .unwrap(); + let worker = dispatched.attempt.worker.as_ref().unwrap(); + seed_session( + &session_root(f.dir.path(), worker), + session_store::new_session_id(), + message("delivered immutable Journal input"), + session_store::LoggedSessionHistoryOrigin::LocalHumanInput, + ); + f.api + .store + .update_worker_retention( + TEST_WORKSPACE_ID, + worker, + "pinned", + &Utc::now().to_rfc3339(), + ) + .unwrap(); + f.api + .finish_backend_job_attempt( + &reserved.job.request.job_id, + &dispatched.attempt.attempt_id, + BackendJobAttemptState::Unknown, + "response_lost", + "test lost result", + ) + .unwrap(); + wait_for_backend_job_worker_cleanup( + &f.api, + &reserved.job.request.job_id, + &dispatched.attempt.attempt_id, + BackendJobWorkerCleanupState::Failed, + ) + .await; + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::OK.as_u16(), + "{}", + response.body + ); + f.assert_removed_with_history(); + assert_eq!( + f.api + .store + .get_backend_job(TEST_WORKSPACE_ID, &reserved.job.request.job_id) + .unwrap() + .unwrap() + .state, + BackendJobState::Unknown + ); + f.api + .store + .update_worker_retention( + TEST_WORKSPACE_ID, + worker, + "normal", + &Utc::now().to_rfc3339(), + ) + .unwrap(); + let attempt = f + .api + .store + .get_backend_job_attempt( + TEST_WORKSPACE_ID, + &reserved.job.request.job_id, + &dispatched.attempt.attempt_id, + ) + .unwrap() + .unwrap(); + f.api.schedule_backend_job_worker_cleanup(&attempt); + wait_for_backend_job_worker_cleanup( + &f.api, + &reserved.job.request.job_id, + &dispatched.attempt.attempt_id, + BackendJobWorkerCleanupState::Completed, + ) + .await; + // Cleanup's own reconciliation owns the CAS; do not race it with a stale snapshot. + tokio::time::timeout(std::time::Duration::from_secs(10), async { + loop { + let job = f + .api + .store + .get_backend_job(TEST_WORKSPACE_ID, &reserved.job.request.job_id) + .unwrap() + .unwrap(); + if job.state == BackendJobState::Failed { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + // Exercise restart recovery against the persisted state while the asynchronous + // cleanup task is still allowed to hold this fixture's Runtime owner lock. + subjektiv_journal::recover(&f.api).unwrap(); + let job = f + .api + .store + .get_backend_job(TEST_WORKSPACE_ID, &reserved.job.request.job_id) + .unwrap() + .unwrap(); + assert_eq!(job.state, BackendJobState::Failed); + assert_eq!(job.current_attempt, 1); + assert_eq!( + job.load_executable_request().unwrap().input, + reserved.job.request.input + ); + assert!(f.store.journal_retention_required(&f.host).unwrap()); +} + +#[tokio::test] +async fn subject_removal_erasure_between_runtime_receipt_and_commit_requires_reconciliation() { + let f = SubjectRemovalFixture::new().await; + let inventory = f.api.runtime.worker_retention_inventory(&f.target).unwrap(); + let plan = f + .api + .config_store + .plan_worker_removal( + &crate::retention::WorkerRemovalPlanRequest { + workspace_id: TEST_WORKSPACE_ID.into(), + worker: f.target.clone(), + reason: "erase race".into(), + journal_erasure: None, + }, + &inventory, + ) + .unwrap(); + let prepared = f + .api + .config_store + .prepare_worker_removal_execution(TEST_WORKSPACE_ID, &plan.plan_id, &plan.input_fingerprint) + .unwrap(); + let service = WorkerRemovalService::new(&f.api); + subjektiv_journal::validate_removal_retention(&service, &f.target, &prepared).unwrap(); + let result = f + .api + .runtime + .execute_worker_retention(&f.target, prepared.runtime_request.clone()) + .unwrap(); + // The domain fence commits, but the original moved to the not-yet-published archive. + assert!( + f.api + .erase_subjektiv_journal_session(&f.attr.session_id, "erase-during-removal") + .is_err() + ); + assert!(f.store.journal_session_erased(&f.host).unwrap()); + let mut committed = false; + assert!( + subjektiv_journal::finish_worker_removal(&service, &f.target, &prepared, &result, || { + committed = true; + }) + .is_err() + ); + assert!(!committed); + f.assert_occupied(); + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::OK.as_u16(), + "{}", + response.body + ); + f.assert_erased_removal(); + assert!( + f.store + .journal_worker_was_released( + &f.attr.subject_id, + &f.target.runtime_id, + &f.target.worker_id + ) + .unwrap() + ); +} + +#[tokio::test] +async fn subject_removal_prior_archive_policy_receipt_remains_recoverable() { + let f = SubjectRemovalFixture::new().await; + f.seed_completed_release_for_archive_recovery(); + let configured = f + .api + .config_store + .worker_retention_policy(TEST_WORKSPACE_ID) + .unwrap() + .unwrap(); + let conn = rusqlite::Connection::open(&f.api.config.database_path).unwrap(); + conn.execute_batch("CREATE TRIGGER interrupt_legacy_removal BEFORE DELETE ON worker_registry BEGIN SELECT RAISE(ABORT,'injected metadata failure'); END;").unwrap(); + assert_ne!(f.remove().status, StatusCode::OK.as_u16()); + let prepared = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + assert_eq!(prepared.plan.policy_id, configured.policy_id); + assert_eq!(prepared.plan.policy_digest, configured.policy_digest); + assert_eq!( + prepared.plan.state, + crate::retention::WorkerRemovalPlanState::Failed + ); + assert!(!session_root(f.dir.path(), &f.target).exists()); + conn.execute_batch("DROP TRIGGER interrupt_legacy_removal;") + .unwrap(); + drop(conn); + let f = f.reopen().await; + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::OK.as_u16(), + "{}", + response.body + ); + f.assert_removed_with_history(); + let recovered = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + assert_eq!(prepared.plan.operation_id, recovered.plan.operation_id); + assert_eq!( + prepared.plan.input_fingerprint, + recovered.plan.input_fingerprint + ); + assert_eq!(prepared.plan.policy_digest, recovered.plan.policy_digest); +} + +async fn assert_subject_removal_recovers_unknown_snapshot(legacy: bool, completed: bool) { + let f = SubjectRemovalFixture::new().await; + let reserved = f.pending_journal_input(legacy); + let grant = reserved + .job + .request + .grants + .subjektiv_journal + .as_ref() + .unwrap(); + if completed { + // Fixture a committed domain checkpoint whose final Job response was lost. + // No Controller/model delivery is claimed by these direct domain writes. + let g = f + .store + .active_journal_generation(&f.host, &grant.journal_id) + .unwrap() + .unwrap(); + let ctx = JournalJobContext::validated(&f.store, &f.host, &g.id, g.job).unwrap(); + let record = f + .store + .execute_journal( + &ctx, + JournalOperation { + request_id: "saved-record".into(), + expected_state: g.initial_journal.state_token, + action: JournalAction::ConfirmUnchanged, + covered_ranges: grant.selected_ranges.clone(), + }, + ) + .unwrap(); + let position = f + .store + .execute_journal( + &ctx, + JournalOperation { + request_id: "saved-position".into(), + expected_state: record.state_token, + action: JournalAction::UpdatePosition { + current_position: "Saved before lost response".into(), + }, + covered_ranges: vec![], + }, + ) + .unwrap(); + f.store + .execute_journal( + &ctx, + JournalOperation { + request_id: "saved-checkpoint".into(), + expected_state: position.state_token, + action: JournalAction::Checkpoint { + record_operation_ids: vec!["saved-record".into()], + }, + covered_ranges: grant.selected_ranges.clone(), + }, + ) + .unwrap(); + subjektiv::journal::service::finish(&f.store, &ctx).unwrap(); + } + // With no bound execution Worker, the store records cleanup Completed. The + // unfinished case must remain Unknown (absence is not a Purge receipt). + f.api + .store + .finish_backend_job_attempt( + TEST_WORKSPACE_ID, + &reserved.job.request.job_id, + &reserved.attempt.attempt_id, + BackendJobAttemptState::Unknown, + "lost-response", + "fixture retained outcome", + &Utc::now().to_rfc3339(), + ) + .unwrap(); + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::OK.as_u16(), + "{}", + response.body + ); + f.assert_removed_with_history(); + let job = f + .api + .store + .get_backend_job(TEST_WORKSPACE_ID, &reserved.job.request.job_id) + .unwrap() + .unwrap(); + assert_eq!( + job.state, + if completed { + BackendJobState::Completed + } else { + BackendJobState::Unknown + } + ); + assert_eq!(job.current_attempt, 1); +} + +#[tokio::test] +async fn subject_removal_recovers_completed_unknown_without_recursive_subject_lock() { + assert_subject_removal_recovers_unknown_snapshot(false, true).await; +} + +#[tokio::test] +async fn subject_removal_recovers_legacy_completed_unknown_without_writer_authority() { + assert_subject_removal_recovers_unknown_snapshot(true, true).await; +} + +#[tokio::test] +async fn subject_removal_retains_legacy_unbound_unknown_without_blocking_deletion() { + assert_subject_removal_recovers_unknown_snapshot(true, false).await; +} + +#[tokio::test] +async fn subject_removal_prior_purge_receipt_recovers_without_inventing_erasure() { + assert_prior_purge_receipt(false).await; +} + +#[tokio::test] +async fn subject_removal_prior_pending_purge_receipt_recovers_with_stale_runtime_catalog() { + assert_prior_purge_receipt(true).await; +} + +async fn assert_prior_purge_receipt(stale_catalog: bool) { + let f = SubjectRemovalFixture::new().await; + f.seed_completed_release_for_archive_recovery(); + let configured = f + .api + .config_store + .worker_retention_policy(TEST_WORKSPACE_ID) + .unwrap() + .unwrap(); + f.api + .config_store + .update_worker_retention_policy( + TEST_WORKSPACE_ID, + &configured.policy_digest, + &crate::retention::WorkerRetentionPolicyUpdate { + policy_id: "old-ordinary-purge".into(), + session_disposition: worker_runtime::retention::SessionDisposition::Purge, + metadata_disposition: configured.metadata_disposition, + archive_retention: configured.archive_retention, + diagnostics_disposition: worker_runtime::retention::DiagnosticsDisposition::Purge, + diagnostics_retention_seconds: None, + }, + ) + .unwrap(); + let lease = f + .api + .config_store + .retained_worker_singleton_lease(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + let conn = rusqlite::Connection::open(&f.api.config.database_path).unwrap(); + // Model the pre-upgrade planner's policy choice only; restore the exact Subject + // reservation before execution. All subsequent retention/receipt paths are real. + conn.execute("UPDATE worker_create_reservations SET singleton_key=NULL,singleton_generation=NULL WHERE workspace_id=?1 AND worker_id=?2", + rusqlite::params![TEST_WORKSPACE_ID, f.target.worker_id]).unwrap(); + let inventory = f.api.runtime.worker_retention_inventory(&f.target).unwrap(); + let plan = f + .api + .config_store + .plan_worker_removal( + &crate::retention::WorkerRemovalPlanRequest { + workspace_id: TEST_WORKSPACE_ID.into(), + worker: f.target.clone(), + reason: "pre-upgrade ordinary Purge".into(), + journal_erasure: None, + }, + &inventory, + ) + .unwrap(); + conn.execute("UPDATE worker_create_reservations SET singleton_key=?3,singleton_generation=?4 WHERE workspace_id=?1 AND worker_id=?2", + rusqlite::params![TEST_WORKSPACE_ID, f.target.worker_id, lease.key, lease.generation]).unwrap(); + drop(conn); + assert_eq!( + plan.session_disposition, + worker_runtime::retention::SessionDisposition::Purge + ); + let prepared = f + .api + .config_store + .prepare_worker_removal_execution(TEST_WORKSPACE_ID, &plan.plan_id, &plan.input_fingerprint) + .unwrap(); + if stale_catalog { + // Fixture the Runtime delete-to-final-receipt crash window. Its live + // catalog deliberately remains stopped while the aggregate is gone. + let request = &prepared.runtime_request; + let pending = worker_runtime::retention::WorkerRetentionExecutionResult { + operation_id: request.operation_id.clone(), + input_fingerprint: request.input_fingerprint.clone(), + worker_id: request.worker_id, + session_disposition: request.session_disposition, + diagnostics_disposition: request.diagnostics_disposition, + archive: None, + source_removed: false, + diagnostics_retained: false, + }; + let receipt = f + .dir + .path() + .join(".test-embedded-runtime-store/retention/operations") + .join(format!("{}.json", plan.operation_id)); + std::fs::create_dir_all(receipt.parent().unwrap()).unwrap(); + std::fs::write( + receipt, + serde_json::to_vec(&serde_json::json!({ + "schema_version":1, "request":request, "result":pending, + })) + .unwrap(), + ) + .unwrap(); + std::fs::remove_dir_all(session_root(f.dir.path(), &f.target).parent().unwrap()).unwrap(); + assert_eq!(f.api.runtime.worker(&f.target).unwrap().state, "stopped"); + } else { + let result = f + .api + .runtime + .execute_worker_retention(&f.target, prepared.runtime_request.clone()) + .unwrap(); + assert!(result.source_removed); + } + f.api + .config_store + .fail_worker_removal( + TEST_WORKSPACE_ID, + &plan.operation_id, + &plan.input_fingerprint, + "metadata_commit_failed", + ) + .unwrap(); + let f = if stale_catalog { f } else { f.reopen().await }; + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::OK.as_u16(), + "{}", + response.body + ); + assert!(!f.store.journal_session_erased(&f.host).unwrap()); + assert!( + f.api + .store + .get_worker_registry(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .is_none() + ); + let recovered = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + assert_eq!(recovered.plan.operation_id, plan.operation_id); + assert_eq!(recovered.plan.input_fingerprint, plan.input_fingerprint); + assert_eq!( + recovered.plan.state, + crate::retention::WorkerRemovalPlanState::Succeeded + ); +} + #[tokio::test] async fn disabled_subject_removal_does_not_admit_new_public_history_or_reserve_a_job() { let f = SubjectRemovalFixture::new().await; @@ -2074,7 +2992,7 @@ async fn disabled_subject_removal_does_not_admit_new_public_history_or_reserve_a let response = f.remove(); assert_eq!( response.status, - StatusCode::CONFLICT.as_u16(), + StatusCode::OK.as_u16(), "{}", response.body ); @@ -2083,7 +3001,7 @@ async fn disabled_subject_removal_does_not_admit_new_public_history_or_reserve_a .store .get_worker_registry(TEST_WORKSPACE_ID, &f.target) .unwrap() - .is_some() + .is_none() ); assert!(f.store.journal_for_session(&f.host).unwrap().is_none()); assert!( @@ -2093,6 +3011,7 @@ async fn disabled_subject_removal_does_not_admit_new_public_history_or_reserve_a .unwrap() .is_empty() ); + f.assert_removed_with_history(); } #[tokio::test] @@ -2148,7 +3067,7 @@ async fn disabled_subject_removal_retains_only_the_already_admitted_prefix() { let response = f.remove(); assert_eq!( response.status, - StatusCode::CONFLICT.as_u16(), + StatusCode::OK.as_u16(), "{}", response.body ); @@ -2157,7 +3076,7 @@ async fn disabled_subject_removal_retains_only_the_already_admitted_prefix() { .store .get_worker_registry(TEST_WORKSPACE_ID, &f.target) .unwrap() - .is_some() + .is_none() ); assert!( f.api @@ -2171,11 +3090,12 @@ async fn disabled_subject_removal_retains_only_the_already_admitted_prefix() { Some(id.as_str()) ); assert!(f.store.journal_retention_required(&f.host).unwrap()); - assert!(session_root(f.dir.path(), &f.target).exists()); + assert!(!session_root(f.dir.path(), &f.target).exists()); + f.assert_removed_with_history(); } #[tokio::test] -async fn attributed_unmaterialized_subject_still_requires_actual_idle_release() { +async fn attributed_unmaterialized_subject_removal_releases_without_restoring_controller() { let f = SubjectRemovalFixture::with_materialized_source(false).await; let retention = source_retention(f.dir.path()); let session = f.attr.session_id.parse().unwrap(); @@ -2200,19 +3120,20 @@ async fn attributed_unmaterialized_subject_still_requires_actual_idle_release() let response = f.remove(); assert_eq!( response.status, - StatusCode::CONFLICT.as_u16(), + StatusCode::OK.as_u16(), "{}", response.body ); - assert_eq!(retention.pins(session).unwrap().len(), 1); + assert!(retention.pins(session).unwrap().is_empty()); assert!( f.api .store .get_worker_registry(TEST_WORKSPACE_ID, &f.target) .unwrap() - .is_some() + .is_none() ); assert!(f.store.journal_for_session(&f.host).unwrap().is_none()); + f.assert_removed_with_history(); } #[tokio::test] @@ -2222,11 +3143,11 @@ async fn missing_materialized_subject_source_blocks_removal_without_marking_it_e let response = f.remove(); assert_eq!( response.status, - StatusCode::CONFLICT.as_u16(), + StatusCode::SERVICE_UNAVAILABLE.as_u16(), "{}", response.body ); - assert!(response.body.contains("journal_handoff_incomplete")); + assert!(response.body.contains("runtime_retention_failed")); assert!( f.api .store @@ -2242,6 +3163,7 @@ async fn missing_materialized_subject_source_blocks_removal_without_marking_it_e .unwrap() .is_empty() ); + f.assert_occupied(); } fn source_retention(root: &std::path::Path) -> session_store::retention::SessionRetentionStore { diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index 702dfb5ac..85e1f24cd 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -3112,8 +3112,8 @@ impl SqliteWorkspaceStore { }, ) .optional()?; - // Subject occupancy survives removal until an explicit release - // proof. It must not resurrect the removed creation reservation: + // An unconfirmed legacy removal cannot release Subject occupancy. + // Completed removal or explicit release does; never resurrect a reservation: // Runtime callbacks require a live reservation or catalog member. if stored.as_ref().is_some_and(|row| row.6 == "removed") { return Err(Error::RepositoryConflict(format!( @@ -16223,7 +16223,7 @@ fn current_worker_singleton_owner( CASE WHEN substr(owner.singleton_key,1,10)='subjektiv:' AND reservation.state='removed' \ AND NOT EXISTS(SELECT 1 FROM subject_worker_releases r WHERE r.workspace_id=owner.workspace_id AND r.worker_id=owner.worker_id AND r.runtime_id=owner.runtime_id AND r.singleton_key=owner.singleton_key AND r.singleton_generation=owner.generation AND r.completed=1) \ AND NOT EXISTS(SELECT 1 FROM subject_create_recoveries r WHERE r.workspace_id=owner.workspace_id AND r.worker_id=owner.worker_id AND r.runtime_id=owner.runtime_id AND r.singleton_key=owner.singleton_key AND r.singleton_generation=owner.generation AND r.create_fingerprint=reservation.create_fingerprint AND r.completed=1) \ - THEN 'created' ELSE reservation.state END \ + AND NOT EXISTS(SELECT 1 FROM worker_removal_operations r WHERE r.workspace_id=owner.workspace_id AND r.worker_id=owner.worker_id AND r.runtime_id=owner.runtime_id AND r.state='succeeded') THEN 'created' ELSE reservation.state END \ FROM worker_singleton_owners owner \ LEFT JOIN worker_create_reservations reservation \ ON reservation.workspace_id = owner.workspace_id \