Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

fix: budget subjektiv batches for compact verified Job results (T-704)

Browse Source
This commit is contained in:
Hare
2026-10-06 19:58:11 +09:00
parent d12cdc9106
commit 3c424fd17d
4 changed files with 233 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/server/subjektiv_job.rs
+66 -8
View File
@@ -147,6 +147,29 @@ pub(super) async fn retire_legacy_worker(api: &WorkspaceApi, subject_id: &str) -
.is_none())
}
fn bounded_candidate_batch(
ids: impl IntoIterator<Item = String>,
max_result_bytes: u32,
) -> Result<Vec<String>> {
// Reserve the bounded subject and Host surface envelope. Each candidate ID
// appears twice (input claims and verified outcome); account for JSON escaping,
// not raw UTF-8 bytes. The fixed per-item allowance covers action and framing.
let mut estimated_bytes = 2_048_usize;
let mut selected = Vec::new();
for id in ids.into_iter().take(CANDIDATE_BATCH_LIMIT) {
let id_bytes = serde_json::to_vec(&id)
.map_err(|e| Error::InvalidInput(e.to_string()))?
.len();
let additional = 2 * id_bytes + 128;
if estimated_bytes + additional > max_result_bytes as usize {
break;
}
estimated_bytes += additional;
selected.push(id);
}
Ok(selected)
}
pub(super) fn dispatch(
api: &WorkspaceApi,
subject_id: &str,
@@ -158,12 +181,14 @@ pub(super) fn dispatch(
.subject(subject_id)
.map_err(subjektiv_store_error)?
.ok_or_else(|| Error::SubjektivSubjectNotFound(subject_id.into()))?;
let candidate_ids: Vec<_> = store
.pending_staging_candidates(subject_id, CANDIDATE_BATCH_LIMIT)
.map_err(subjektiv_store_error)?
.into_iter()
.map(|c| c.id)
.collect();
let candidate_ids = bounded_candidate_batch(
store
.pending_staging_candidates(subject_id, CANDIDATE_BATCH_LIMIT)
.map_err(subjektiv_store_error)?
.into_iter()
.map(|c| c.id),
crate::backend_job::DEFAULT_MAX_RESULT_BYTES,
)?;
let batch_len = candidate_ids.len();
let request = BackendJobRequest {
job_id: format!("subjektiv-consolidation:{}", Uuid::new_v4()),
@@ -296,8 +321,10 @@ pub(super) fn with_dispositions(
.ok_or_else(|| {
Error::InvalidInput(format!("candidate `{id}` has no durable disposition"))
})?;
dispositions
.push(serde_json::to_value(resolution).map_err(|e| Error::Store(e.to_string()))?);
// Full reasons, affected revisions and source edges remain in the existing
// immutable receipt store. Duplicating them makes a valid bounded batch's
// completion unbounded; this result is its verified compact projection.
dispositions.push(serde_json::json!({ "candidate_id": id, "action": resolution.action }));
}
let mut enriched = value.clone();
enriched
@@ -309,3 +336,34 @@ pub(super) fn with_dispositions(
);
Ok(enriched)
}
#[cfg(test)]
mod result_budget_tests {
use super::*;
#[test]
fn escaped_maximum_ids_leave_a_bounded_complete_result() {
let ids: Vec<_> = (0..100)
.map(|i| format!("{i:03}{}", "\\\"".repeat(126)))
.collect();
let batch =
bounded_candidate_batch(ids.clone(), crate::backend_job::DEFAULT_MAX_RESULT_BYTES)
.unwrap();
assert!(!batch.is_empty());
assert!(batch.len() < ids.len());
assert_eq!(batch, ids[..batch.len()]);
let dispositions: Vec<_> = batch
.iter()
.map(|id| serde_json::json!({"candidate_id":id,"action":"already_covered"}))
.collect();
let result = serde_json::json!({
"subject_id":"\\\"".repeat(128),"candidate_ids":batch,
"candidate_dispositions":dispositions,
"surface":{"availability":"failed","generation_id":"g".repeat(100),"store_revision":u64::MAX,"reason_code":"\\\"".repeat(128)},
});
assert!(
serde_json::to_vec(&result).unwrap().len()
<= crate::backend_job::DEFAULT_MAX_RESULT_BYTES as usize
);
}
}
crates/workspace-server/src/server/tests/subjektiv_jobs_tests.rs
+123
View File
@@ -607,3 +607,126 @@ async fn subjektiv_job_legacy_cutover_waits_foreground_then_removes_only_owned_c
assert!(api.runtime.worker(&subject_body).is_ok());
assert!(store.subject(&subject.id).unwrap().is_some());
}
#[tokio::test]
async fn subjektiv_job_long_receipts_and_escaped_ids_fit_result_budget_without_losing_followup() {
let dir = tempfile::tempdir().unwrap();
let (api, _) = test_api_with_recording_backend(dir.path()).await;
let store = open_subjektiv_store(&api).unwrap();
let subject = store
.create_subject(SubjectRole::new("result-budget").unwrap())
.unwrap();
for i in 0..100 {
stage(
&store,
&subject.id,
&format!("candidate-{i:03}-{}", "\\\"".repeat(60)),
);
}
start_subjektiv_staging_consolidation(
api.clone(),
&subject.id,
MemoryConsolidateStagingOperation { force: true },
)
.await
.unwrap();
let job = resource(&api, &subject.id);
let ids = job
.request
.grants
.subjektiv_consolidation
.as_ref()
.unwrap()
.candidate_ids
.clone();
assert!(!ids.is_empty() && ids.len() < 100);
let attempt = current_attempt(&api, &job);
let worker = attempt.worker.as_ref().unwrap();
for id in &ids {
store
.decide_candidate(
&subject.id,
CandidateDecisionRequest {
request_id: format!("decision-{}", uuid::Uuid::new_v4()),
candidate_id: id.clone(),
reason: "Existing confirmed detail remains in its immutable receipt. "
.repeat(500),
decision: CandidateDecision::Close {
action: crate::subjektiv::StagingResolutionAction::Discarded,
affected_memory: vec![],
},
},
)
.unwrap();
}
let generation = store
.prepare_job_surface_generation(
&subject.id,
Some((&job.request.job_id, &attempt.attempt_id)),
)
.unwrap();
let snapshot = store
.publish_surface_generation(&subject.id, &generation.id, vec![])
.unwrap();
let submission = BackendJobResultSubmission {
job_id: job.request.job_id.clone(),
attempt_id: attempt.attempt_id.clone(),
input_revision: job.request.input_revision.clone(),
result: serde_json::json!({"subject_id":subject.id,"candidate_ids":ids,"surface":{"availability":"ready","generation_id":generation.id,"store_revision":generation.store_revision,"snapshot_id":snapshot.id}}),
};
seed_worker_session_for_cleanup(dir.path(), worker);
let accepted = api.accept_backend_job_result(worker, &submission).unwrap();
let result = accepted.job.result.as_ref().unwrap();
assert!(
serde_json::to_vec(result).unwrap().len() <= job.request.limits.max_result_bytes as usize
);
let outcomes = result["candidate_dispositions"].as_array().unwrap();
assert_eq!(outcomes.len(), ids.len());
assert!(
outcomes
.iter()
.all(|o| o["action"] == "discarded" && o.get("reason").is_none())
);
assert_eq!(
store.pending_staging_backlog(&subject.id).unwrap().0,
100 - ids.len()
);
assert!(
api.accept_backend_job_result(worker, &submission)
.unwrap()
.replayed
);
wait_for_backend_job_worker_cleanup(
&api,
&job.request.job_id,
&attempt.attempt_id,
BackendJobWorkerCleanupState::Completed,
)
.await;
start_subjektiv_staging_consolidation(
api.clone(),
&subject.id,
MemoryConsolidateStagingOperation { force: true },
)
.await
.unwrap();
let next = resource(&api, &subject.id);
assert!(
next.request
.grants
.subjektiv_consolidation
.unwrap()
.candidate_ids
.iter()
.all(|id| !ids.contains(id))
);
assert!(
store
.staging_resolution(&subject.id, &ids[0])
.unwrap()
.unwrap()
.reason
.len()
> 16 * 1024
);
}