fix(worker): reconcile invalid Drive completions and default pages

Accept the Backend shared default page maxima (list 200/search 128), not an unsent 100 limit. Classify semantically invalid completions before receipt reconciliation; query the original identity once and accept only validated committed results, never resend or upgrade revisions.

Regressions fail on 61afdb77 and pass after the fix: real API 201-entry default Tools/native WIP pagination; mismatched completion ID/invalid entry metadata recovery and unresolved/invalid receipt matrices. Root cargo check; Worker 920 unit +110 integration; Server 709 lib +11 binary; Drive unit18/API18; fmt and diff checks passed.
This commit is contained in:
2026-10-09 15:36:06 +09:00
parent 61afdb7773
commit e94af40dab
4 changed files with 251 additions and 21 deletions
@@ -172,7 +172,12 @@ impl DriveBackend {
}, },
) )
.await?; .await?;
self.validate_page(&result, limit.unwrap_or(100).min(DRIVE_PAGE_MAX_LIMIT))?; self.validate_page(
&result,
limit
.unwrap_or(DRIVE_PAGE_MAX_LIMIT)
.min(DRIVE_PAGE_MAX_LIMIT),
)?;
if result if result
.entries .entries
.iter() .iter()
@@ -186,7 +191,10 @@ impl DriveBackend {
let result: DriveListResponse = self.get("/search", &query).await?; let result: DriveListResponse = self.get("/search", &query).await?;
self.validate_page( self.validate_page(
&result, &result,
query.limit.unwrap_or(100).min(DRIVE_SEARCH_MAX_LIMIT), query
.limit
.unwrap_or(DRIVE_SEARCH_MAX_LIMIT)
.min(DRIVE_SEARCH_MAX_LIMIT),
)?; )?;
Ok(result) Ok(result)
} }
@@ -276,25 +284,37 @@ impl DriveBackend {
id: &str, id: &str,
result: Result<DriveMutationResponse, DriveError>, result: Result<DriveMutationResponse, DriveError>,
) -> Result<DriveMutationResponse, DriveError> { ) -> Result<DriveMutationResponse, DriveError> {
let response = match result { // A parseable success can still be an invalid completion. Classify it
// before reconciliation so it also gets one original-request inquiry.
let result = result.and_then(|response| {
self.validate_completion(id, &response)
.map_err(|_| DriveError::OutcomeUnknown(id.into()))?;
Ok(response)
});
match result {
Err(DriveError::OutcomeUnknown(_)) => match self.status(id).await { Err(DriveError::OutcomeUnknown(_)) => match self.status(id).await {
Ok(DriveRequestStatusResponse { Ok(DriveRequestStatusResponse {
state: DriveRequestState::Committed, state: DriveRequestState::Committed,
response: Some(response), response: Some(response),
.. ..
}) => response, }) => Ok(response), // status validates both IDs and entry metadata
_ => return Err(DriveError::OutcomeUnknown(id.into())), _ => Err(DriveError::OutcomeUnknown(id.into())),
}, },
other => other?, other => other,
}; }
}
fn validate_completion(
&self,
id: &str,
response: &DriveMutationResponse,
) -> Result<(), DriveError> {
if response.request_id != id { if response.request_id != id {
return Err(DriveError::OutcomeUnknown(id.into())); return Err(DriveError::Unavailable);
} }
if let Some(entry) = &response.entry { if let Some(entry) = &response.entry {
self.validate_entry(entry) self.validate_entry(entry)?;
.map_err(|_| DriveError::OutcomeUnknown(id.into()))?;
} }
Ok(response) Ok(())
} }
pub async fn mutate( pub async fn mutate(
&self, &self,
@@ -371,12 +391,7 @@ impl DriveBackend {
return Err(DriveError::Unavailable); return Err(DriveError::Unavailable);
} }
if let Some(result) = &response.response { if let Some(result) = &response.response {
if result.request_id != request_id { self.validate_completion(request_id, result)?;
return Err(DriveError::Unavailable);
}
if let Some(entry) = &result.entry {
self.validate_entry(entry)?;
}
} }
Ok(response) Ok(response)
} }
@@ -294,6 +294,143 @@ async fn uncommitted_receipt_snapshot_does_not_turn_unknown_into_failure_or_retr
assert_eq!(client.requests.lock().unwrap().len(), 2); assert_eq!(client.requests.lock().unwrap().len(), 2);
} }
fn invalid_completions(id: &str) -> Vec<DriveMutationResponse> {
let mut invalid_entry = entry("2", "2");
invalid_entry.revision = "not-a-revision".into();
vec![
DriveMutationResponse {
request_id: "another-request".into(),
entry: Some(entry("2", "2")),
},
DriveMutationResponse {
request_id: id.into(),
entry: Some(invalid_entry),
},
]
}
#[tokio::test]
async fn invalid_mutation_completion_recovers_from_original_receipt_without_resending() {
let context = ToolExecutionContext::new("write", "batch", 0);
let client = Client::new(vec![]);
let feature = feature(client.clone());
let id = feature.request_id(&context);
for completion in invalid_completions(&id) {
let committed = DriveMutationResponse {
request_id: id.clone(),
entry: Some(entry("2", "2")),
};
client.requests.lock().unwrap().clear();
client.responses.lock().unwrap().extend([
Ok(response(completion)),
Ok(response(DriveRequestStatusResponse {
request_id: id.clone(),
state: DriveRequestState::Committed,
response: Some(committed.clone()),
})),
]);
feature.observe(entry("2", "1")).unwrap();
let output = feature
.execute(
"DriveWrite",
&json!({"entry":entry("2", "1").entry,"content":"new"}).to_string(),
&context,
None,
)
.await
.unwrap();
assert_eq!(output.value["request_id"], id);
assert_eq!(output.value["entry"]["metadata"]["revision"], "2");
let requests = client.requests.lock().unwrap();
assert_eq!(requests.len(), 2);
assert_eq!(
requests[0].method,
crate::worker::WorkspaceRequestMethod::Post
);
assert_eq!(
requests[1].method,
crate::worker::WorkspaceRequestMethod::Get
);
assert!(requests[1].path.ends_with(&format!("/requests/{id}")));
let body: Value = serde_json::from_str(requests[0].body.as_ref().unwrap()).unwrap();
assert_eq!(body["request_id"], id);
assert_eq!(body["mutation"]["expected_revision"], "1");
}
}
#[tokio::test]
async fn invalid_completion_and_unresolved_or_invalid_receipt_stay_unknown_after_one_query() {
let context = ToolExecutionContext::new("write", "batch", 0);
let client = Client::new(vec![]);
let feature = feature(client.clone());
let id = feature.request_id(&context);
let committed = DriveMutationResponse {
request_id: id.clone(),
entry: Some(entry("2", "2")),
};
let mut receipts = vec![
response(DriveRequestStatusResponse {
request_id: id.clone(),
state: DriveRequestState::Uncommitted,
response: None,
}),
response(DriveRequestStatusResponse {
request_id: id.clone(),
state: DriveRequestState::Committed,
response: None,
}),
response(DriveRequestStatusResponse {
request_id: "another-request".into(),
state: DriveRequestState::Committed,
response: Some(committed),
}),
WorkspaceResponse {
status: 200,
body: "invalid-json".into(),
},
];
receipts.extend(invalid_completions(&id).into_iter().map(|completion| {
response(DriveRequestStatusResponse {
request_id: id.clone(),
state: DriveRequestState::Committed,
response: Some(completion),
})
}));
for completion in invalid_completions(&id) {
for receipt in &receipts {
client.requests.lock().unwrap().clear();
client
.responses
.lock()
.unwrap()
.extend([Ok(response(&completion)), Ok(receipt.clone())]);
feature.observe(entry("2", "1")).unwrap();
let result = feature
.execute(
"DriveWrite",
&json!({"entry":entry("2", "1").entry,"content":"new"}).to_string(),
&context,
None,
)
.await;
assert!(
matches!(result, Err(DriveError::OutcomeUnknown(ref observed)) if observed == &id)
);
let requests = client.requests.lock().unwrap();
assert_eq!(requests.len(), 2);
assert_eq!(
requests[0].method,
crate::worker::WorkspaceRequestMethod::Post
);
assert_eq!(
requests[1].method,
crate::worker::WorkspaceRequestMethod::Get
);
assert!(requests[1].path.ends_with(&format!("/requests/{id}")));
}
}
}
#[tokio::test] #[tokio::test]
async fn observation_eviction_never_substitutes_another_entry() { async fn observation_eviction_never_substitutes_another_entry() {
let feature = feature(Client::new(Vec::new())); let feature = feature(Client::new(Vec::new()));
@@ -726,6 +726,80 @@ async fn unsafe_names_and_oversized_text_cannot_publish_and_reads_truncate_at_ut
} }
} }
#[tokio::test]
async fn omitted_discovery_limits_use_backend_maximum_pages_in_tools_and_native_wip() {
let f = Fixture::new(["read_write", "read_only"]).await;
let mut expected = std::collections::BTreeSet::new();
for index in 0..201 {
let created = f
.create(&format!("default-page-{index:03}.md"), "bounded")
.await;
expected.insert(entry(&created)["node_id"].as_str().unwrap().to_owned());
}
let native = native_tools(&f.features[1]);
let root = value(
native_tool(&native, "Inspect", json!({"path":"/drive","refresh":true}))
.await
.unwrap(),
);
let reference = root["interfaces"][0]["reference"].clone();
for use_native in [false, true] {
for (name, operation, mut args, maximum) in [
(
"DriveList",
"list",
json!({}),
server_api::DRIVE_PAGE_MAX_LIMIT as usize,
),
(
"DriveSearch",
"search",
json!({"query":"default-page-"}),
server_api::DRIVE_SEARCH_MAX_LIMIT as usize,
),
] {
let mut actual = Vec::new();
for page_index in 0..2 {
let output = if use_native {
native_tool(&native, "Invoke", json!({"path":"/drive","interface":reference,"operation":operation,"arguments":args})).await
} else {
tool(&f.features[1], name, args.clone()).await
};
let page = value(output.unwrap());
let entries = page["entries"].as_array().unwrap();
assert_eq!(
entries.len(),
if page_index == 0 {
maximum
} else {
201 - maximum
},
"native={use_native} operation={operation}"
);
actual.extend(entries.iter().map(|e| {
e["metadata"]["entry"]["node_id"]
.as_str()
.unwrap()
.to_owned()
}));
if page_index == 0 {
assert!(page["next_after"].is_string());
args["after"] = page["next_after"].clone();
} else {
assert!(page["next_after"].is_null());
}
}
assert_eq!(actual.len(), expected.len());
assert_eq!(
actual
.into_iter()
.collect::<std::collections::BTreeSet<_>>(),
expected
);
}
}
}
#[tokio::test] #[tokio::test]
async fn list_and_search_pages_are_bounded_disjoint_and_cursors_follow_backend_scope() { async fn list_and_search_pages_are_bounded_disjoint_and_cursors_follow_backend_scope() {
let f = Fixture::new(["read_write", "read_only"]).await; let f = Fixture::new(["read_write", "read_only"]).await;
+8 -4
View File
@@ -63,7 +63,9 @@ authenticated latest URLs. Deleting/recreating the same name allocates a new ID.
| `DriveSaveWorkdir` | explicit `target_workdir` alias, relative `path`, destination `parent`, `name`, optional `content_type` (octet-stream by default) | | `DriveSaveWorkdir` | explicit `target_workdir` alias, relative `path`, destination `parent`, `name`, optional `content_type` (octet-stream by default) |
| `DriveRequestStatus` | `request_id` from an uncertain mutation | | `DriveRequestStatus` | `request_id` from an uncertain mutation |
List/search are paged discovery only, not implicit mutation observations. Read or List/search are paged discovery only, not implicit mutation observations. Omitting
`limit` uses the Backend maximum (200 for list, 128 for search) in both Tools and
native WIP; use the returned `next_after` to continue. Read or
metadata-inspect an entry before replacing, editing, moving or deleting it. The metadata-inspect an entry before replacing, editing, moving or deleting it. The
Tools adapter retains at most 256 observations, scoped to its injected Workspace. Tools adapter retains at most 256 observations, scoped to its injected Workspace.
Eviction requires a new explicit observation; it never substitutes another entry. Eviction requires a new explicit observation; it never substitutes another entry.
@@ -120,9 +122,11 @@ latest and retry. Observe explicitly, reconsider the edit and submit a deliberat
new operation. Native Client stale handling likewise does not replay dispatch. new operation. Native Client stale handling likewise does not replay dispatch.
Timeout, lost response, oversized/invalid mutation response or invalid completion Timeout, lost response, oversized/invalid mutation response or invalid completion
may mean **OutcomeUnknown**. The adapter queries the *same* request receipt once; may mean **OutcomeUnknown**, including a completion with a mismatched request ID
if committed it returns that confirmed DB result. Otherwise it preserves unknown or invalid entry metadata. The adapter queries the *same* request receipt once;
and the `request_id`. Query `DriveRequestStatus`; an `uncommitted` snapshot is not only a committed receipt with matching IDs and valid metadata returns confirmed
DB success. Otherwise it preserves unknown and the `request_id`. Query
`DriveRequestStatus`; an `uncommitted` snapshot is not
proof that an in-flight request cannot later commit. Do not retry blindly with a proof that an in-flight request cannot later commit. Do not retry blindly with a
fresh request ID. Backend grants also protect receipt lookup/replay. Safe typed fresh request ID. Backend grants also protect receipt lookup/replay. Safe typed
Denied/NotFound/Conflict/Invalid/Limit/StorageUnavailable/OutcomeUnknown errors are Denied/NotFound/Conflict/Invalid/Limit/StorageUnavailable/OutcomeUnknown errors are