diff --git a/crates/nokv-client/src/artifact.rs b/crates/nokv-client/src/artifact.rs index 300a9334..f0faf308 100644 --- a/crates/nokv-client/src/artifact.rs +++ b/crates/nokv-client/src/artifact.rs @@ -2801,6 +2801,42 @@ mod tests { assert!(state.attempts.is_empty()); } + #[test] + fn valid_index_fields_cross_the_workspace_rpc_publication_path() { + let events = Arc::new(Mutex::new(Vec::new())); + let transport = ScriptedArtifactTransport::new(Arc::clone(&events), false); + let inspector = transport.clone(); + let client = client(transport); + let store = RecordingStore::new(events, None); + let index_fields = vec![ + FieldValue { + field_id: "agent.owner".to_owned(), + value: ScalarValue::String("planner".to_owned()), + }, + FieldValue { + field_id: "agent.score".to_owned(), + value: ScalarValue::Unsigned(7), + }, + ]; + + client + .publish_artifact( + &store, + publish_options(4).with_index_fields(index_fields.clone()), + b"abcdefgh", + ) + .unwrap(); + + let state = inspector.state(); + let metadata = state + .path + .as_ref() + .and_then(|path| path.metadata.as_ref()) + .expect("publication exposes the visible path metadata"); + assert_eq!(metadata.descriptor.index_fields, index_fields); + assert!(state.attempts.iter().any(|(label, _)| *label == "complete")); + } + #[test] fn response_loss_replays_the_exact_stage_request_without_duplicate_apply() { let events = Arc::new(Mutex::new(Vec::new())); diff --git a/crates/nokv-server/src/executor.rs b/crates/nokv-server/src/executor.rs index 9a10817b..c60e6de1 100644 --- a/crates/nokv-server/src/executor.rs +++ b/crates/nokv-server/src/executor.rs @@ -1343,7 +1343,7 @@ impl MetadataWorkspaceRequestExecutor { ) -> Result { self.claim_mutation(rpc)?; let transition_id = derived_request_id(rpc.request_id, b"publish-complete-transition", 0); - let _transition = if let Some(outcome) = self.replayed_publish(rpc.route, transition_id)? { + let transition = if let Some(outcome) = self.replayed_publish(rpc.route, transition_id)? { outcome } else { let operation = self.heartbeat_publish_operation( @@ -1372,9 +1372,13 @@ impl MetadataWorkspaceRequestExecutor { true, ); } + let finalizing_token = protocol::OperationToken { + operation_id: transition.operation.operation_id.into(), + state_digest: publish_state_digest(&transition.operation)?, + }; let finalizing_operation = self.heartbeat_publish_operation( rpc, - request.token, + finalizing_token, b"publish-heartbeat-complete-finalize", )?; let context = self.publication_context(rpc.route, finalize_id)?; @@ -4958,17 +4962,81 @@ mod tests { } #[test] - fn search_crosses_the_real_query_facade_and_decodes_typed_indexes() { - let (store, executor) = ready_executor(); - let incarnation = types::WorkspaceIncarnationId::from_bytes([41; types::FIXED_ID_BYTES]); + fn published_index_fields_cross_workspace_rpc_and_query_facade() { + let (_store, executor) = ready_executor(); executor .execute(&create_request(40, "query-test", 41, 1)) .unwrap(); - put_visible_path(&store, incarnation); + let operation_id = protocol::OperationIdentity([0x51; types::FIXED_ID_BYTES]); + let artifact_revision_id = + protocol::ArtifactRevisionIdentity([0x52; types::FIXED_ID_BYTES]); + let target = protocol::WorkspacePath { + workbench: protocol::WorkbenchName::new("query-test").unwrap(), + path: protocol::RelativePath::new("outputs/result.bin").unwrap(), + }; + let index_fields = vec![protocol::FieldValue { + field_id: "agent.score".to_owned(), + value: protocol::ScalarValue::Unsigned(7), + }]; + let seals = protocol::seal_artifact_publish_plan(artifact_revision_id, &[], &[]).unwrap(); + let begun = executor + .execute(&protocol::WorkspaceRpcRequest { + route: route(1), + request_id: protocol::RequestIdentity([0x53; types::FIXED_ID_BYTES]), + operation: protocol::WorkspaceRequest::BeginArtifactPublish( + protocol::BeginArtifactPublishRequest { + operation_id, + artifact_revision_id, + target: target.clone(), + authority: protocol::PublicationAuthority::Visible, + condition: protocol::PublishCondition::CreateOnly, + staged_object_count: seals.staged_object_count, + staged_object_seal: seals.staged_object_seal, + manifest_row_count: seals.manifest_row_count, + manifest_seal: seals.manifest_seal, + dependency_owner_revision_ids: Vec::new(), + }, + ), + }) + .unwrap(); + let protocol::WorkspaceResult::Operation(status) = begun.result else { + panic!("begin artifact publication returned the wrong result variant"); + }; + assert_eq!(status.state, protocol::OperationState::Running); + assert_eq!(status.progress.completed_rows, 0); + assert_eq!(status.progress.total_rows, Some(0)); + + let completed = executor + .execute(&protocol::WorkspaceRpcRequest { + route: route(1), + request_id: protocol::RequestIdentity([0x54; types::FIXED_ID_BYTES]), + operation: protocol::WorkspaceRequest::CompleteArtifactPublish( + protocol::CompleteArtifactPublishRequest { + token: status.token, + artifact: protocol::ArtifactDescriptor { + logical_size: 0, + body_digest: protocol::sha256_digest_uri(protocol::Digest( + Sha256::digest([]).into(), + )), + manifest_digest: protocol::sha256_digest_uri(seals.manifest_seal), + content_type: protocol::ContentType::new("application/octet-stream") + .unwrap(), + producer: Some("executor-test".to_owned()), + manifest_identity: Some("manifest-1".to_owned()), + index_fields: index_fields.clone(), + }, + }, + ), + }) + .unwrap(); + let protocol::WorkspaceResult::Published(published) = completed.result else { + panic!("complete artifact publication returned the wrong result variant"); + }; + assert_eq!(published.target, target); let search = protocol::WorkspaceRpcRequest { route: route(1), - request_id: protocol::RequestIdentity([42; types::FIXED_ID_BYTES]), + request_id: protocol::RequestIdentity([0x55; types::FIXED_ID_BYTES]), operation: protocol::WorkspaceRequest::Search(protocol::SearchRequest { scope: protocol::QueryScope::Root { path_prefix: None }, predicates: vec![protocol::QueryPredicate { @@ -4993,12 +5061,10 @@ mod tests { panic!("search returned the wrong result variant"); }; assert_eq!(result.hits.len(), 1); + assert_eq!(result.hits[0].projection, index_fields); assert_eq!( - result.hits[0].projection, - vec![protocol::FieldValue { - field_id: "agent.score".to_owned(), - value: protocol::ScalarValue::Unsigned(7), - }] + result.hits[0].metadata.descriptor.index_fields, + index_fields ); assert_eq!( result.hits[0].metadata.path.path.as_str(), @@ -5009,7 +5075,7 @@ mod tests { let catalog = protocol::WorkspaceRpcRequest { route: route(1), - request_id: protocol::RequestIdentity([43; types::FIXED_ID_BYTES]), + request_id: protocol::RequestIdentity([0x56; types::FIXED_ID_BYTES]), operation: protocol::WorkspaceRequest::Catalog(protocol::CatalogRequest { scope: protocol::QueryScope::Workspace { workbench: protocol::WorkbenchName::new("query-test").unwrap(), @@ -5032,11 +5098,11 @@ mod tests { .expect("one-row catalog page has a continuation"); executor - .execute(&create_request(44, "query-version-advance", 45, 1)) + .execute(&create_request(0x57, "query-version-advance", 0x58, 1)) .unwrap(); let stale_catalog = protocol::WorkspaceRpcRequest { route: route(1), - request_id: protocol::RequestIdentity([46; types::FIXED_ID_BYTES]), + request_id: protocol::RequestIdentity([0x59; types::FIXED_ID_BYTES]), operation: protocol::WorkspaceRequest::Catalog(protocol::CatalogRequest { scope: protocol::QueryScope::Workspace { workbench: protocol::WorkbenchName::new("query-test").unwrap(), diff --git a/docs/metadata-schema.md b/docs/metadata-schema.md index 663436bc..088d939b 100644 --- a/docs/metadata-schema.md +++ b/docs/metadata-schema.md @@ -401,6 +401,12 @@ typed index projection There is no canonical inode, dentry, parent pointer, link count, directory record, or fallback path index. +Clients attach queryable fields to `ArtifactDescriptor.index_fields` on the +artifact publication request. The workspace executor validates and encodes +that typed projection, and final publication updates `PathCurrent` and the +corresponding `SecondaryIndex` rows in the same metadata command. There is no +separate namespace-index registration RPC or second mutation path. + A cold exact artifact lookup performs one `WorkspaceCurrent` point read and, only when its state is `Visible`, one `PathCurrent` point read. `PathEntry` atomically retains the immutable revision fields required to shape complete