diff --git a/crates/tracedecay-application/src/advisory/github_runtime/network.rs b/crates/tracedecay-application/src/advisory/github_runtime/network.rs index 38f84787ee2..49a63ff3ae0 100644 --- a/crates/tracedecay-application/src/advisory/github_runtime/network.rs +++ b/crates/tracedecay-application/src/advisory/github_runtime/network.rs @@ -2029,7 +2029,7 @@ mod pagination_contract_tests { } #[cfg(test)] -mod tests { +pub(super) mod tests { use std::collections::BTreeSet; use std::io::{Read, Write}; use std::net::TcpListener; @@ -2066,7 +2066,7 @@ mod tests { include_str!("../fixtures/provider_branch_review/review_thread.graphql.json"); #[derive(Clone, Copy)] - pub(super) enum FixtureCredentialAuthorityModeV1 { + pub(in crate::advisory::github_runtime) enum FixtureCredentialAuthorityModeV1 { Verified, NotConfigured, WriteCapable, @@ -2139,7 +2139,7 @@ mod tests { } } - pub(super) fn registered_fixture_credential( + pub(in crate::advisory::github_runtime) fn registered_fixture_credential( repository: &str, mode: FixtureCredentialAuthorityModeV1, ) -> ( @@ -2716,7 +2716,7 @@ mod tests { assert_eq!(outcome, GitHubReadNetworkOutcomeV1::Denied); } - pub(super) fn scope(suffix: &str) -> FeedbackScopeV1 { + pub(in crate::advisory::github_runtime) fn scope(suffix: &str) -> FeedbackScopeV1 { FeedbackScopeV1 { project_id: ProjectId::new(format!("project.github.{suffix}")).unwrap(), repository_id: RepositoryId::new(format!("repository.github.{suffix}")).unwrap(), @@ -2726,7 +2726,7 @@ mod tests { } } - pub(super) fn context(scope: &FeedbackScopeV1) -> RequestContext { + pub(in crate::advisory::github_runtime) fn context(scope: &FeedbackScopeV1) -> RequestContext { let resolved = ResolvedScope::new( scope.project_id.clone(), scope.repository_id.clone(), @@ -2763,7 +2763,9 @@ mod tests { .unwrap() } - pub(super) fn request(scope: FeedbackScopeV1) -> GitHubReviewReadRequestV1 { + pub(in crate::advisory::github_runtime) fn request( + scope: FeedbackScopeV1, + ) -> GitHubReviewReadRequestV1 { GitHubReviewReadRequestV1 { operation: GitHubReviewReadOperationV1::GraphQlQueryPullRequestReviewThreads, scope, diff --git a/crates/tracedecay-application/src/advisory/github_runtime/network/stack_network.rs b/crates/tracedecay-application/src/advisory/github_runtime/network/stack_network.rs index 9ae957c2cca..aef768ebab9 100644 --- a/crates/tracedecay-application/src/advisory/github_runtime/network/stack_network.rs +++ b/crates/tracedecay-application/src/advisory/github_runtime/network/stack_network.rs @@ -143,7 +143,8 @@ impl GitHubReadOnlyClientV1 { #[cfg(test)] mod tests { use std::io::Write; - use std::net::TcpListener; + use std::net::{SocketAddr, TcpListener}; + use std::sync::mpsc; use serde_json::json; use tracedecay_contracts::feedback::{FeedbackPortFuture, GitHubReviewReadRequestV1}; @@ -151,9 +152,10 @@ mod tests { GitTopologyAnchorAuthority, GitTopologyAnchorResolution, GitTopologyAnchorResolutionOutcome, }; use tracedecay_contracts::{RequestContext, now_micros}; + use tracedecay_domain::configuration::GitHubStackedPullRequestPolicyV1; use tracedecay_domain::{ - CommitId, GitTopologyAnchorTargetV1, ObservationScopeV1, ProviderId, RetrievalAnchorTarget, - UtcMicros, + CommitId, GitHubStackCapabilityStateV1, GitTopologyAnchorTargetV1, ObservationScopeV1, + ProviderId, RetrievalAnchorTarget, UtcMicros, }; use tracedecay_global_db::tests::harness::open_registered_test_database_fixture; use tracedecay_runtime_core::db::TestDatabaseRuntimeScope; @@ -165,7 +167,8 @@ mod tests { FixtureCredentialAuthorityModeV1, context, registered_fixture_credential, request, scope, }; use super::super::{ - GitHubHttpReadConfigV1, GitHubRepositoryTargetV1, RegisteredGitHubReadOnlyCredentialV1, + GitHubHttpReadConfigV1, GitHubReadOnlyCredentialV1, GitHubRepositoryTargetV1, + RegisteredGitHubReadOnlyCredentialV1, }; use super::*; use crate::advisory::github_runtime::{ @@ -173,7 +176,69 @@ mod tests { GitHubSourceAccessAuthorityV1, GitHubStackAnchorReadOutcomeV1, ProjectGitHubStackAnchorAuthorityV1, }; - use crate::stack_coordinator::{DaemonGitHubStackCoordinatorV1, GitHubStackProviderSnapshotV1}; + use crate::stack_coordinator::{ + DaemonGitHubStackCoordinatorV1, GitHubStackCoordinatorErrorV1, + GitHubStackProviderSnapshotV1, + }; + + const MERGE_BASE_COMMIT: &str = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; + + fn private_stack_response(head_commit: &str, head_ref_name: &str) -> serde_json::Value { + json!({ + "data": { "repository": { "pullRequest": { + "stackEntry": { "position": 1 }, + "stack": { + "id": "stack-node-private-1", + "number": 421, + "baseRefName": "main", + "size": 1, + "entries": { + "totalCount": 1, + "pageInfo": { "hasNextPage": false, "endCursor": null }, + "nodes": [{ + "position": 1, + "pullRequest": { + "number": 421, + "baseRefName": "main", + "headRefName": head_ref_name, + "baseRefOid": MERGE_BASE_COMMIT, + "headRefOid": head_commit, + "baseRef": null, + "statusCheckRollup": { "state": "SUCCESS" }, + "mergeQueueEntry": { + "id": "queue-private-1", + "position": 1, + "state": "QUEUED" + } + } + }] + } + } + }}} + }) + } + + fn merge_base_response() -> serde_json::Value { + json!({ "merge_base_commit": { "sha": MERGE_BASE_COMMIT } }) + } + + fn local_stack_client( + address: SocketAddr, + repository: &str, + request: &GitHubReviewReadRequestV1, + credential: GitHubReadOnlyCredentialV1, + ) -> GitHubReadOnlyClientV1 { + super::super::test_support::local_client( + address, + GitHubRepositoryTargetV1 { + owner: "ScriptedAlchemy".to_owned(), + repository: repository.to_owned(), + pull_request_number: 421, + pull_request_id: request.pull_request_id.clone(), + }, + credential, + ) + } struct ReadyStackSourceAccess; @@ -341,38 +406,7 @@ mod tests { assert!(payload["query"].as_str().unwrap().contains("stackEntry")); write_http_json( &mut graphql, - &json!({ - "data": { "repository": { "pullRequest": { - "stackEntry": { "position": 1 }, - "stack": { - "id": "stack-node-private-1", - "number": 421, - "baseRefName": "main", - "size": 1, - "entries": { - "totalCount": 1, - "pageInfo": { "hasNextPage": false, "endCursor": null }, - "nodes": [{ - "position": 1, - "pullRequest": { - "number": 421, - "baseRefName": "main", - "headRefName": "github-stack-anchor-http", - "baseRefOid": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", - "headRefOid": head_commit, - "baseRef": null, - "statusCheckRollup": { "state": "SUCCESS" }, - "mergeQueueEntry": { - "id": "queue-private-1", - "position": 1, - "state": "QUEUED" - } - } - }] - } - } - }}} - }), + &private_stack_response(&head_commit, "github-stack-anchor-http"), ); let (mut compare, _) = listener.accept().unwrap(); let (headers, _) = read_http_request_with_headers(&mut compare); @@ -382,33 +416,20 @@ mod tests { .contains("authorization: bearer github_pat_fixture_private_read") ); assert!(headers.contains("/compare/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa...")); - write_http_json( - &mut compare, - &json!({ - "merge_base_commit": { "sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" } - }), - ); + write_http_json(&mut compare, &merge_base_response()); }); - let client = GitHubReadOnlyClientV1 { - agent: crate::http_agent::http_agent( - ureq::Agent::config_builder() - .https_only(false) - .http_status_as_error(false) - .build(), - ), - target: GitHubRepositoryTargetV1 { - owner: "ScriptedAlchemy".to_owned(), - repository: "stack-anchor-http".to_owned(), - pull_request_number: 421, - pull_request_id: request.pull_request_id.clone(), - }, - credential, - config: GitHubHttpReadConfigV1 { - rest_base_uri: format!("http://{address}"), - graphql_uri: format!("http://{address}/graphql"), - ..GitHubHttpReadConfigV1::default() - }, - }; + let client = local_stack_client(address, "stack-anchor-http", &request, credential); + let coordinator = DaemonGitHubStackCoordinatorV1::default(); + coordinator + .register_scope( + context.scope(), + GitHubStackedPullRequestPolicyV1::ProbePrivatePreview, + ) + .unwrap(); + let stack_read = coordinator + .authorize_provider_read(context.scope()) + .unwrap() + .expect("a probing registration authorizes the read"); let provider_outcome = client.read_stack( &context, &GitHubGraphQlReadRequestV1 { @@ -436,16 +457,9 @@ mod tests { let source_binding = anchors .source_binding(&context, &request, &provider_outcome, observed_at) .unwrap(); - let coordinator = DaemonGitHubStackCoordinatorV1::default(); - coordinator - .register_scope( - context.scope(), - tracedecay_domain::configuration::GitHubStackedPullRequestPolicyV1::ProbePrivatePreview, - ) - .unwrap(); let observation = coordinator .observe_provider( - context.scope().clone(), + stack_read, provider, provider_outcome, source_binding.clone(), @@ -538,4 +552,143 @@ mod tests { } } } + + /// The read is authorized before its network I/O. Disabling and + /// re-enabling the scope while the provider response is held leaves the + /// policy equal again, yet the late completion is refused; only a read + /// authorized by the new registration publishes. + #[tokio::test] + async fn stack_read_held_across_policy_reregistration_cannot_publish() { + let mut scope = scope("stack-held-read"); + scope.head_commit_id = CommitId::new("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb").unwrap(); + let context = context(&scope); + let request = request(scope.clone()); + let storage = tempfile::tempdir().unwrap(); + let (database, _owner) = open_registered_test_database_fixture( + &storage.path().join("project-sessions.db"), + TestDatabaseRuntimeScope::ProjectSessions { + project_id: scope.project_id.clone(), + }, + ) + .await + .unwrap(); + let anchors = + ProjectGitHubStackAnchorAuthorityV1::new(database.clone(), scope.clone()).unwrap(); + let provider = ProviderId::new("provider.github").unwrap(); + let (_credential_authority, resolution) = registered_fixture_credential( + "stack-held-read", + FixtureCredentialAuthorityModeV1::Verified, + ); + let RegisteredGitHubReadOnlyCredentialV1::Verified(credential) = resolution else { + panic!("verified fixture credential must resolve"); + }; + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let head_commit = scope.head_commit_id.as_str().to_owned(); + let (arrived, request_arrived) = mpsc::channel(); + let (release, released) = mpsc::channel::<()>(); + let server = std::thread::spawn(move || { + for held in [true, false] { + let (mut graphql, _) = listener.accept().unwrap(); + let _ = read_http_request(&mut graphql); + if held { + arrived.send(()).unwrap(); + released.recv().unwrap(); + } + write_http_json( + &mut graphql, + &private_stack_response(&head_commit, "github-stack-held-read"), + ); + let (mut compare, _) = listener.accept().unwrap(); + let _ = read_http_request(&mut compare); + write_http_json(&mut compare, &merge_base_response()); + } + }); + let client = local_stack_client(address, "stack-held-read", &request, credential); + let stack_request = GitHubGraphQlReadRequestV1 { + scope: scope.clone(), + pull_request_id: request.pull_request_id.clone(), + resume: GitHubReadResumeV1::empty(), + }; + let coordinator = DaemonGitHubStackCoordinatorV1::default(); + coordinator + .register_scope( + context.scope(), + GitHubStackedPullRequestPolicyV1::ProbePrivatePreview, + ) + .unwrap(); + + let held_read = coordinator + .authorize_provider_read(context.scope()) + .unwrap() + .expect("a probing registration authorizes the read"); + let in_flight = { + let (client, context, stack_request, request, provider, anchors) = ( + client.clone(), + context.clone(), + stack_request.clone(), + request.clone(), + provider.clone(), + anchors.clone(), + ); + std::thread::spawn(move || { + client.read_stack(&context, &stack_request, &request, &provider, &anchors) + }) + }; + request_arrived.recv().unwrap(); + coordinator + .register_scope(context.scope(), GitHubStackedPullRequestPolicyV1::Disabled) + .unwrap(); + coordinator + .register_scope( + context.scope(), + GitHubStackedPullRequestPolicyV1::ProbePrivatePreview, + ) + .unwrap(); + release.send(()).unwrap(); + let held_outcome = in_flight.join().unwrap(); + assert!( + matches!(held_outcome, GitHubStackProviderOutcomeV1::Enabled(_)), + "the held provider read still returns its stack" + ); + let held_at = now_micros(); + let held_binding = anchors + .source_binding(&context, &request, &held_outcome, held_at) + .unwrap(); + assert_eq!( + coordinator.observe_provider( + held_read, + provider.clone(), + held_outcome, + held_binding, + held_at, + ), + Err(GitHubStackCoordinatorErrorV1::Superseded) + ); + + let fresh_read = coordinator + .authorize_provider_read(context.scope()) + .unwrap() + .expect("the new registration authorizes a fresh read"); + let fresh_outcome = + client.read_stack(&context, &stack_request, &request, &provider, &anchors); + server.join().unwrap(); + let fresh_at = now_micros(); + let fresh_binding = anchors + .source_binding(&context, &request, &fresh_outcome, fresh_at) + .unwrap(); + let observation = coordinator + .observe_provider(fresh_read, provider, fresh_outcome, fresh_binding, fresh_at) + .unwrap(); + assert_eq!( + observation.capability.state, + GitHubStackCapabilityStateV1::Enabled + ); + assert_eq!(observation.observed_at, fresh_at); + assert!(observation.snapshot.is_some()); + assert!( + observation.transitions.is_empty(), + "the new registration has no baseline from the retired one" + ); + } } diff --git a/crates/tracedecay-application/src/advisory/github_runtime/network/test_support.rs b/crates/tracedecay-application/src/advisory/github_runtime/network/test_support.rs index 41fb82d0fc5..29e31cfc204 100644 --- a/crates/tracedecay-application/src/advisory/github_runtime/network/test_support.rs +++ b/crates/tracedecay-application/src/advisory/github_runtime/network/test_support.rs @@ -1,5 +1,10 @@ use std::io::{Read, Write}; -use std::net::TcpStream; +use std::net::{SocketAddr, TcpStream}; + +use super::{ + GitHubHttpReadConfigV1, GitHubReadOnlyClientV1, GitHubReadOnlyCredentialV1, + GitHubRepositoryTargetV1, +}; pub(in crate::advisory::github_runtime) fn read_http_request_with_headers( stream: &mut TcpStream, @@ -83,3 +88,26 @@ pub(in crate::advisory::github_runtime) fn write_http_response( stream.write_all(head.as_bytes()).unwrap(); stream.write_all(&body).unwrap(); } + +/// Real HTTP transport for isolated loopback provider fixtures. +pub(in crate::advisory::github_runtime) fn local_client( + address: SocketAddr, + target: GitHubRepositoryTargetV1, + credential: GitHubReadOnlyCredentialV1, +) -> GitHubReadOnlyClientV1 { + GitHubReadOnlyClientV1 { + agent: crate::http_agent::http_agent( + ureq::Agent::config_builder() + .https_only(false) + .http_status_as_error(false) + .build(), + ), + target, + credential, + config: GitHubHttpReadConfigV1 { + rest_base_uri: format!("http://{address}"), + graphql_uri: format!("http://{address}/graphql"), + ..GitHubHttpReadConfigV1::default() + }, + } +} diff --git a/crates/tracedecay-application/src/advisory/github_runtime/owner.rs b/crates/tracedecay-application/src/advisory/github_runtime/owner.rs index 62a445b10ed..8350a01a728 100644 --- a/crates/tracedecay-application/src/advisory/github_runtime/owner.rs +++ b/crates/tracedecay-application/src/advisory/github_runtime/owner.rs @@ -27,7 +27,8 @@ use crate::observability::{ record_github_stack_drifts, }; use crate::stack_coordinator::{ - DaemonGitHubStackCoordinatorV1, GitHubStackObservationV1, GitHubStackProviderSourceBindingV1, + DaemonGitHubStackCoordinatorV1, GitHubStackCoordinatorErrorV1, GitHubStackObservationV1, + GitHubStackProviderOutcomeV1, GitHubStackProviderSourceBindingV1, }; use tracedecay_global_db::RegisteredGlobalDbLeaseV1; use tracedecay_runtime_core::db::Database; @@ -151,21 +152,24 @@ where { return outcome; } - let stack_read_enabled = match self + // The read is authorized by the scope's registration before any + // network I/O; its completion may publish only under that same + // registration. + let stack_read = match self .stack_coordinator - .should_read_provider_stack(&self.stack_scope) + .authorize_provider_read(&self.stack_scope) { - Ok(enabled) => enabled, + Ok(read) => read, Err(error) => { tracing::warn!( event = "github_stack_policy_read_failed", error = ?error, - "GitHub stack policy observation failed" + "GitHub stack read authorization failed" ); - false + return outcome; } }; - if stack_read_enabled { + if let Some(stack_read) = stack_read { let client = self.stack_client.clone(); let stack_context = context.clone(); let stack_review_request = request.clone(); @@ -211,35 +215,27 @@ where return outcome; }; match self.stack_coordinator.observe_provider( - self.stack_scope.clone(), + stack_read, self.stack_provider.clone(), provider_outcome, source_binding.clone(), stack_observed_at, ) { Ok(observation) => { - // Offered before anchor publication so a publication - // refusal below cannot conceal the observation the - // coordinator already made. - if let Some(stack_observability) = &self.stack_observability { - stack_observability.record(&source_binding, &observation); - } - let anchor_publication = self - .stack_anchors - .publish(context, request, &observation, self.source_access.as_ref()) - .await; - if !matches!( - anchor_publication, - super::GitHubStackAnchorPublicationOutcomeV1::Published - | super::GitHubStackAnchorPublicationOutcomeV1::Replayed - ) { - tracing::warn!( - event = "github_stack_anchor_publication_failed", - outcome = ?anchor_publication, - "GitHub stack evidence was not published to the retrieval-anchor authority" - ); - return outcome; - } + self.publish_stack_observation( + context, + request, + &source_binding, + &observation, + ) + .await; + } + Err(GitHubStackCoordinatorErrorV1::Superseded) => { + tracing::debug!( + event = "github_stack_provider_read_superseded", + "GitHub stack read outlived its scope registration and was not published" + ); + self.observe_stack_policy(context, request).await; } Err(error) => { tracing::warn!( @@ -249,11 +245,97 @@ where ); } } + } else { + self.observe_stack_policy(context, request).await; } outcome }) } + async fn observe_stack_policy( + &self, + context: &RequestContext, + request: &GitHubReviewReadRequestV1, + ) { + // A superseded provider response has no authority over the current + // policy. Bind a policy-only observation to the current source instead. + let observed_at = tracedecay_contracts::now_micros(); + let Some(binding) = self.stack_anchors.source_binding( + context, + request, + &GitHubStackProviderOutcomeV1::Unavailable, + observed_at, + ) else { + tracing::warn!( + event = "github_stack_source_binding_failed", + "GitHub stack policy could not be bound to its exact source owner" + ); + return; + }; + match self.stack_coordinator.observe_policy( + self.stack_scope.clone(), + self.stack_provider.clone(), + binding.clone(), + observed_at, + ) { + Ok(observation) => { + // The coordinator preserves a provider observation already + // published by the current registration; it needs no new + // policy receipt with a different source binding. + if matches!( + observation.capability.state, + tracedecay_domain::GitHubStackCapabilityStateV1::PrivatePreviewDisabled + | tracedecay_domain::GitHubStackCapabilityStateV1::Unavailable + ) { + self.publish_stack_observation(context, request, &binding, &observation) + .await; + } + } + Err(GitHubStackCoordinatorErrorV1::Superseded) => { + tracing::debug!( + event = "github_stack_policy_observation_superseded", + "GitHub stack policy changed before its observation was published" + ); + } + Err(error) => { + tracing::warn!( + event = "github_stack_policy_observation_failed", + error = ?error, + "GitHub stack policy observation failed" + ); + } + } + } + + async fn publish_stack_observation( + &self, + context: &RequestContext, + request: &GitHubReviewReadRequestV1, + binding: &GitHubStackProviderSourceBindingV1, + observation: &GitHubStackObservationV1, + ) { + // Offered before anchor publication so a publication refusal cannot + // conceal the observation the coordinator already made. + if let Some(observability) = &self.stack_observability { + observability.record(binding, observation); + } + let publication = self + .stack_anchors + .publish(context, request, observation, self.source_access.as_ref()) + .await; + if !matches!( + publication, + super::GitHubStackAnchorPublicationOutcomeV1::Published + | super::GitHubStackAnchorPublicationOutcomeV1::Replayed + ) { + tracing::warn!( + event = "github_stack_anchor_publication_failed", + outcome = ?publication, + "GitHub stack evidence was not published to the retrieval-anchor authority" + ); + } + } + pub fn authorize<'a>( &'a self, context: &'a RequestContext, @@ -340,3 +422,312 @@ fn map_admission_error( ) -> GitHubReviewRuntimeOwnerBuildErrorV1 { GitHubReviewRuntimeOwnerBuildErrorV1::InvalidDescriptor } + +#[cfg(test)] +mod tests { + use std::net::TcpListener; + use std::sync::{Arc, mpsc}; + + use serde_json::json; + use tracedecay_domain::configuration::GitHubStackedPullRequestPolicyV1; + use tracedecay_domain::feedback::{ + GitHubReviewCurrentBranchRemapV1, GitHubReviewImmutableAnchorV1, + }; + use tracedecay_domain::{ + CommitId, GitHubStackCapabilityStateV1, GitTopologyAnchorTargetV1, IntervalStateV1, + ProviderId, RetrievalAnchorRecord, RetrievalAnchorTarget, + }; + use tracedecay_global_db::tests::harness::open_registered_test_database_fixture; + use tracedecay_runtime_core::db::{ + DatabaseAuthority, TestDatabaseRuntimeMode, TestDatabaseRuntimeScope, + }; + + use super::super::decoder::{GitHubCanonicalReviewAnchorsV1, GitHubReviewAnchorSeedV1}; + use super::super::network::RegisteredGitHubReadOnlyCredentialV1; + use super::super::network::test_support::{ + local_client, read_http_request_with_headers, write_http_json, + }; + use super::super::network::tests::{ + FixtureCredentialAuthorityModeV1, context, registered_fixture_credential, request, scope, + }; + use super::super::{GitHubProviderLifecycleV1, ProjectGitHubStackAnchorAuthorityV1}; + use super::*; + + #[derive(Clone)] + struct EmptyReviewAnchors; + + impl GitHubCanonicalReviewAnchorAuthorityV1 for EmptyReviewAnchors { + fn resolve<'a>( + &'a self, + _request: &'a GitHubReviewReadRequestV1, + _seed: &'a GitHubReviewAnchorSeedV1, + ) -> FeedbackPortFuture<'a, Option> { + panic!("an empty review collection has no anchors to resolve") + } + } + + impl GitHubCurrentBranchRemapper for EmptyReviewAnchors { + fn remap<'a>( + &'a self, + _context: &'a RequestContext, + _scope: &'a FeedbackScopeV1, + _original: &'a GitHubReviewImmutableAnchorV1, + ) -> FeedbackPortFuture<'a, Option> { + panic!("an empty review collection has no anchors to remap") + } + } + + struct ReadySource; + + impl GitHubSourceAccessAuthorityV1 for ReadySource { + fn authorize<'a>( + &'a self, + _context: &'a RequestContext, + _request: &'a GitHubReviewReadRequestV1, + ) -> FeedbackPortFuture<'a, GitHubProviderLifecycleV1> { + Box::pin(async { GitHubProviderLifecycleV1::Ready }) + } + } + + #[tokio::test] + async fn disabled_refresh_closes_provider_drift_without_a_stack_read() { + policy_disable_journey(false).await; + } + + #[tokio::test] + async fn superseded_refresh_closes_provider_drift_under_current_policy() { + policy_disable_journey(true).await; + } + + async fn policy_disable_journey(hold_read: bool) { + let repository = if hold_read { + "runtime-policy-held" + } else { + "runtime-policy-disabled" + }; + let mut scope = scope(repository); + scope.head_commit_id = CommitId::new("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb").unwrap(); + let context = context(&scope); + let mut request = request(scope.clone()); + request.operation = GitHubReviewReadOperationV1::RestListPullRequestReviewComments; + let storage = tempfile::tempdir().unwrap(); + let (anchor_database, _anchor_owner) = open_registered_test_database_fixture( + &storage.path().join("project-sessions.db"), + TestDatabaseRuntimeScope::ProjectSessions { + project_id: scope.project_id.clone(), + }, + ) + .await + .unwrap(); + crate::register_test_schema_installer(); + let review_path = storage.path().join("review.db"); + let authority = + DatabaseAuthority::acquire_test(&review_path, "github-runtime-policy").unwrap(); + let (review_database, _) = Database::publish_test_runtime( + &review_path, + &authority, + TestDatabaseRuntimeMode::Initialize, + ) + .await + .unwrap(); + let coordinator = Arc::new(DaemonGitHubStackCoordinatorV1::default()); + coordinator + .register_scope( + context.scope(), + GitHubStackedPullRequestPolicyV1::ProbePrivatePreview, + ) + .unwrap(); + let (_credential_authority, resolution) = + registered_fixture_credential(repository, FixtureCredentialAuthorityModeV1::Verified); + let RegisteredGitHubReadOnlyCredentialV1::Verified(credential) = resolution else { + panic!("verified fixture credential must resolve"); + }; + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let (arrived, request_arrived) = tokio::sync::oneshot::channel(); + let (release, released) = mpsc::channel::<()>(); + let head = scope.head_commit_id.as_str().to_owned(); + let branch = scope + .branch_ref + .trim_start_matches("refs/heads/") + .to_owned(); + let server = std::thread::spawn(move || { + let mut arrived = Some(arrived); + for round in 0..3 { + // Complete collection refreshes require two agreeing scans. + for _ in 0..2 { + let (mut review, _) = listener.accept().unwrap(); + let (headers, _) = read_http_request_with_headers(&mut review); + assert!(headers.starts_with("GET ")); + write_http_json(&mut review, &json!([])); + } + if round == 2 && !hold_read { + continue; + } + let (mut stack, _) = listener.accept().unwrap(); + let (headers, body) = read_http_request_with_headers(&mut stack); + assert!( + body["query"] + .as_str() + .is_some_and(|query| query.contains("stackEntry")), + "expected stack query, got {headers} {body}" + ); + if round == 2 { + arrived.take().unwrap().send(()).unwrap(); + released.recv().unwrap(); + } + let base = if round == 0 { + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + } else { + "cccccccccccccccccccccccccccccccccccccccc" + }; + write_http_json( + &mut stack, + &json!({"data":{"repository":{"pullRequest":{ + "stackEntry":{"position":1}, + "stack":{"id":"runtime-stack","number":421,"baseRefName":"main","size":1, + "entries":{"totalCount":1,"pageInfo":{"hasNextPage":false,"endCursor":null}, + "nodes":[{"position":1,"pullRequest":{"number":421,"baseRefName":"main", + "headRefName":branch,"baseRefOid":base,"headRefOid":head,"baseRef":null, + "statusCheckRollup":{"state":"SUCCESS"},"mergeQueueEntry":null}}]}} + }}}}), + ); + let (mut compare, _) = listener.accept().unwrap(); + let _ = read_http_request_with_headers(&mut compare); + write_http_json( + &mut compare, + &json!({"merge_base_commit":{"sha":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}}), + ); + } + }); + let client = local_client( + address, + GitHubRepositoryTargetV1 { + owner: "ScriptedAlchemy".to_owned(), + repository: repository.to_owned(), + pull_request_number: 421, + pull_request_id: request.pull_request_id.clone(), + }, + credential, + ); + let store = ProjectGitHubReviewStoreV1::new(review_database, scope.clone()).unwrap(); + let decoder = GitHubOfficialResponseDecoderV1::new( + GitHubReviewProviderIdentityV1 { + provider: ProviderId::new("provider.github").unwrap(), + repository_owner: "ScriptedAlchemy".to_owned(), + repository_name: repository.to_owned(), + pull_request_number: 421, + base_commit_id: CommitId::new("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa").unwrap(), + head_commit_id: scope.head_commit_id.clone(), + merge_base_commit_id: CommitId::new("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa") + .unwrap(), + }, + EmptyReviewAnchors, + ) + .unwrap(); + let transport = + GitHubReadOnlyRuntimeTransportV1::new(store.clone(), client.clone(), decoder); + let connector = GitHubReadOnlyConnector::new( + GitHubReadOnlyDescriptorSetV1::new(vec![rest_descriptor(request.operation)]).unwrap(), + transport, + EmptyReviewAnchors, + ) + .unwrap(); + let source_access: Arc = Arc::new(ReadySource); + let anchors = + ProjectGitHubStackAnchorAuthorityV1::new(anchor_database.clone(), scope).unwrap(); + let owner = Arc::new(GitHubReviewRuntimeOwnerV1 { + coordinator: GitHubReviewRefreshCoordinatorV1::new( + connector, + store, + Arc::clone(&source_access), + ), + source_access, + stack_client: client, + stack_scope: context.scope().clone(), + stack_provider: ProviderId::new("provider.github").unwrap(), + stack_coordinator: Arc::clone(&coordinator), + stack_anchors: anchors.clone(), + stack_observability: None, + }); + for _ in 0..2 { + let outcome = owner.refresh(&context, &request).await; + assert!(outcome.has_usable_observation(), "{outcome:?}"); + } + let binding = anchors + .source_binding( + &context, + &request, + &GitHubStackProviderOutcomeV1::Unavailable, + tracedecay_contracts::now_micros(), + ) + .unwrap(); + let open = coordinator + .observe_policy( + context.scope().clone(), + owner.stack_provider.clone(), + binding.clone(), + tracedecay_contracts::now_micros(), + ) + .unwrap(); + assert_eq!(open.capability.state, GitHubStackCapabilityStateV1::Enabled); + assert!( + open.drift_observations + .iter() + .any(|drift| drift.drift.state == IntervalStateV1::Open) + ); + if hold_read { + let (owner, context, request) = (Arc::clone(&owner), context.clone(), request.clone()); + let refresh = tokio::spawn(async move { owner.refresh(&context, &request).await }); + request_arrived.await.unwrap(); + coordinator + .register_scope(&open.scope, GitHubStackedPullRequestPolicyV1::Disabled) + .unwrap(); + release.send(()).unwrap(); + assert!(refresh.await.unwrap().has_usable_observation()); + } else { + coordinator + .register_scope(context.scope(), GitHubStackedPullRequestPolicyV1::Disabled) + .unwrap(); + let outcome = owner.refresh(&context, &request).await; + assert!(outcome.has_usable_observation(), "{outcome:?}"); + } + server.join().unwrap(); + let mut records = anchor_database + .read_connection() + .query("SELECT anchor_json FROM retrieval_anchors", ()) + .await + .unwrap(); + let mut published_disabled = false; + while let Some(row) = records.next().await.unwrap() { + let record: RetrievalAnchorRecord = + serde_json::from_str(&row.get::(0).unwrap()).unwrap(); + if matches!(record.target(), RetrievalAnchorTarget::GitTopology(target) + if matches!(target.as_ref(), GitTopologyAnchorTargetV1::GitHubStackCapability(capability) + if capability.state == GitHubStackCapabilityStateV1::PrivatePreviewDisabled)) + { + published_disabled = true; + } + } + assert!( + published_disabled, + "disabled policy must be published to the real anchor store" + ); + let settled = coordinator + .observe_policy( + context.scope().clone(), + owner.stack_provider.clone(), + binding, + tracedecay_contracts::now_micros(), + ) + .unwrap(); + assert_eq!( + settled.capability.state, + GitHubStackCapabilityStateV1::PrivatePreviewDisabled + ); + assert!( + settled.drift_observations.is_empty(), + "the runtime refresh must already have closed provider drift" + ); + } +} diff --git a/crates/tracedecay-application/src/stack_coordinator.rs b/crates/tracedecay-application/src/stack_coordinator.rs index bfb28bc1718..b986c50d5b3 100644 --- a/crates/tracedecay-application/src/stack_coordinator.rs +++ b/crates/tracedecay-application/src/stack_coordinator.rs @@ -184,6 +184,9 @@ impl GitHubStackDriftObservationV1 { pub enum GitHubStackCoordinatorErrorV1 { InvalidScope, InvalidProviderObservation, + /// The scope registration that authorized the observation was replaced + /// before it could publish. + Superseded, Poisoned, } @@ -197,10 +200,83 @@ pub enum StackCoordinatorErrorV1 { Invalid(String), } +/// Identity of one scope registration. Every holder keeps the allocation +/// alive, so a successor registration can never compare equal to a retired +/// one that is still referenced. +#[derive(Clone, Debug)] +struct ScopeRegistrationId(Arc<()>); + +impl ScopeRegistrationId { + fn new() -> Self { + Self(Arc::new(())) + } +} + +fn same_registration( + left: Option<&ScopeRegistrationId>, + right: Option<&ScopeRegistrationId>, +) -> bool { + match (left, right) { + (Some(left), Some(right)) => Arc::ptr_eq(&left.0, &right.0), + (None, None) => true, + _ => false, + } +} + +#[derive(Clone, Debug)] +struct ScopeRegistration { + id: ScopeRegistrationId, + policy: GitHubStackedPullRequestPolicyV1, +} + +/// One provider stack read authorized by the scope's current registration. +/// +/// Capture it before network I/O and surrender it to +/// [`DaemonGitHubStackCoordinatorV1::observe_provider`] on completion. +/// Publication is refused unless that same registration is still current, so +/// a read spanning a policy change, including a change and its reversal, +/// cannot publish. Dropping it abandons the read without touching state. +#[derive(Debug)] +pub struct GitHubStackProviderReadV1 { + scope: ResolvedScope, + registration: ScopeRegistrationId, +} + +/// Who derived an observation and under which registration. +enum ObservationOrigin { + /// Derived from the registered policy alone; it never replaces a provider + /// result published under the same registration. + Policy(Option), + /// A provider read authorized by this registration. + Provider(ScopeRegistrationId), +} + +impl ObservationOrigin { + fn registration(&self) -> Option<&ScopeRegistrationId> { + match self { + Self::Policy(registration) => registration.as_ref(), + Self::Provider(registration) => Some(registration), + } + } +} + +struct PublishedObservation { + registration: Option, + observation: GitHubStackObservationV1, +} + +/// A scope's registration, latest observation and open drift intervals +/// advance together under one lock. +#[derive(Default)] +struct ScopeObservationState { + registration: Option, + current: Option, + drift_intervals: Vec, +} + #[derive(Default)] pub struct DaemonGitHubStackCoordinatorV1 { - policies: Mutex>, - observations: Mutex>, + observation_states: Mutex>, circuit_policies: Mutex>, /// Held across the append so a duplicate debounced signal racing the /// first one cannot append twice. @@ -209,7 +285,6 @@ pub struct DaemonGitHubStackCoordinatorV1 { >, circuits: Mutex>, preflights: Mutex, - drift_intervals: Mutex>>, } impl DaemonGitHubStackCoordinatorV1 { @@ -221,25 +296,59 @@ impl DaemonGitHubStackCoordinatorV1 { scope .validate() .map_err(|_| GitHubStackCoordinatorErrorV1::InvalidScope)?; - self.policies + let mut states = self + .observation_states .lock() - .map_err(|_| GitHubStackCoordinatorErrorV1::Poisoned)? - .insert(scope.scope_digest.clone(), policy); + .map_err(|_| GitHubStackCoordinatorErrorV1::Poisoned)?; + let state = states.entry(scope.scope_digest.clone()).or_default(); + // A changed policy is a new registration: reads it authorized can no + // longer publish, and its snapshot stops being a drift baseline while + // its timestamp still fences late samples. Re-registering an unchanged + // policy keeps the registration and its in-flight reads. + if state + .registration + .as_ref() + .is_none_or(|registration| registration.policy != policy) + { + state.registration = Some(ScopeRegistration { + id: ScopeRegistrationId::new(), + policy, + }); + } Ok(()) } - pub fn should_read_provider_stack( + /// Authorizes one provider stack read when the scope's current + /// registration probes the private preview; `None` means no read may run. + pub fn authorize_provider_read( + &self, + scope: &ResolvedScope, + ) -> Result, GitHubStackCoordinatorErrorV1> { + Ok(self + .registration(scope)? + .filter(|registration| { + registration.policy == GitHubStackedPullRequestPolicyV1::ProbePrivatePreview + }) + .map(|registration| GitHubStackProviderReadV1 { + scope: scope.clone(), + registration: registration.id, + })) + } + + fn registration( &self, scope: &ResolvedScope, - ) -> Result { - let policies = self - .policies + ) -> Result, GitHubStackCoordinatorErrorV1> { + scope + .validate() + .map_err(|_| GitHubStackCoordinatorErrorV1::InvalidScope)?; + let states = self + .observation_states .lock() .map_err(|_| GitHubStackCoordinatorErrorV1::Poisoned)?; - Ok(matches!( - policies.get(&scope.scope_digest), - Some(GitHubStackedPullRequestPolicyV1::ProbePrivatePreview) - )) + Ok(states + .get(&scope.scope_digest) + .and_then(|state| state.registration.clone())) } pub fn observe_policy( @@ -249,28 +358,14 @@ impl DaemonGitHubStackCoordinatorV1 { source_binding: GitHubStackProviderSourceBindingV1, observed_at: UtcMicros, ) -> Result { - let state = if self.should_read_provider_stack(&scope)? { + let registration = self.registration(&scope)?; + let state = if registration.as_ref().is_some_and(|registration| { + registration.policy == GitHubStackedPullRequestPolicyV1::ProbePrivatePreview + }) { GitHubStackCapabilityStateV1::Unavailable } else { GitHubStackCapabilityStateV1::PrivatePreviewDisabled }; - if state == GitHubStackCapabilityStateV1::Unavailable - && let Some(provider_observation) = self - .observations - .lock() - .map_err(|_| GitHubStackCoordinatorErrorV1::Poisoned)? - .get(&scope.scope_digest) - .filter(|observation| { - matches!( - observation.capability.state, - GitHubStackCapabilityStateV1::Enabled - | GitHubStackCapabilityStateV1::Degraded - ) - }) - .cloned() - { - return Ok(provider_observation); - } let digest = canonical_sha256(&( "tracedecay.github-stack.policy-observation.v1", &scope.scope_digest, @@ -288,20 +383,24 @@ impl DaemonGitHubStackCoordinatorV1 { source_binding: &source_binding, }, observed_at, + ObservationOrigin::Policy(registration.map(|registration| registration.id)), ) } + /// Publishes the completion of `read`. A read whose registration was + /// replaced while it ran is refused as `Superseded` and changes nothing. pub fn observe_provider( &self, - scope: ResolvedScope, + read: GitHubStackProviderReadV1, provider: ProviderId, outcome: GitHubStackProviderOutcomeV1, source_binding: GitHubStackProviderSourceBindingV1, observed_at: UtcMicros, ) -> Result { - if !self.should_read_provider_stack(&scope)? { - return self.observe_policy(scope, provider, source_binding, observed_at); - } + let GitHubStackProviderReadV1 { + scope, + registration, + } = read; let state = outcome.capability_state(); let response_digest = match outcome.response_digest() { Some(digest) => digest.clone(), @@ -326,6 +425,7 @@ impl DaemonGitHubStackCoordinatorV1 { source_binding: &source_binding, }, observed_at, + ObservationOrigin::Provider(registration), ) } @@ -348,7 +448,24 @@ impl DaemonGitHubStackCoordinatorV1 { scope: &ResolvedScope, observation: &GitHubStackDriftObservationV1, ) -> Result<(), GitHubStackCoordinatorErrorV1> { - transition::restore_open_drift_interval(&self.drift_intervals, scope, observation) + let interval = transition::restored_open_interval(scope, observation)?; + let mut states = self + .observation_states + .lock() + .map_err(|_| GitHubStackCoordinatorErrorV1::Poisoned)?; + let active = &mut states + .entry(scope.scope_digest.clone()) + .or_default() + .drift_intervals; + if let Some(existing) = active + .iter_mut() + .find(|existing| existing.kind == interval.kind) + { + *existing = interval; + } else { + active.push(interval); + } + Ok(()) } #[tracing::instrument(name = "usecases.stack.enqueue", level = "trace", skip_all)] @@ -608,6 +725,7 @@ impl DaemonGitHubStackCoordinatorV1 { state: GitHubStackCapabilityStateV1, material: GitHubStackObservationMaterial<'_>, observed_at: UtcMicros, + origin: ObservationOrigin, ) -> Result { scope .validate() @@ -689,16 +807,59 @@ impl DaemonGitHubStackCoordinatorV1 { }) .transpose() .map_err(|_| GitHubStackCoordinatorErrorV1::InvalidProviderObservation)?; - let previous = self - .observations + // Serialize comparison, interval update, and snapshot publication under + // one scope-state transition. No network or storage I/O occurs while held. + let mut states = self + .observation_states .lock() - .map_err(|_| GitHubStackCoordinatorErrorV1::Poisoned)? - .get(&scope.scope_digest) - .and_then(|observation| observation.snapshot.as_ref()) - .cloned(); - let (transitions, drift_kinds) = stack_drift(previous.as_ref(), snapshot.as_ref()); + .map_err(|_| GitHubStackCoordinatorErrorV1::Poisoned)?; + let state = states.entry(scope.scope_digest.clone()).or_default(); + // The observation was derived under the registration it names. Policy + // and provider authority both end when that registration is replaced. + if !same_registration( + state + .registration + .as_ref() + .map(|registration| ®istration.id), + origin.registration(), + ) { + return Err(GitHubStackCoordinatorErrorV1::Superseded); + } + // Only an observation published under this same registration is a + // drift baseline or a provider result worth preserving. + let registered_current = state.current.as_ref().filter(|published| { + same_registration(published.registration.as_ref(), origin.registration()) + }); + // A policy observation prepared before a real provider result of the + // same registration must not overwrite its enabled/degraded result. + if let ObservationOrigin::Policy(_) = origin + && let Some(previous) = registered_current.filter(|published| { + matches!( + published.observation.capability.state, + GitHubStackCapabilityStateV1::Enabled | GitHubStackCapabilityStateV1::Degraded + ) + }) + { + return Ok(previous.observation.clone()); + } + // Provider responses may finish out of order. A late sample cannot + // replace a newer snapshot or close a restored interval before it opened. + if state + .current + .as_ref() + .is_some_and(|current| current.observation.observed_at > observed_at) + || state + .drift_intervals + .iter() + .any(|interval| interval.first_observed_at > observed_at) + { + return Err(GitHubStackCoordinatorErrorV1::InvalidProviderObservation); + } + let previous = + registered_current.and_then(|published| published.observation.snapshot.as_ref()); + let (transitions, drift_kinds) = stack_drift(previous, snapshot.as_ref()); let drift_observations = transition::update_drift_intervals( - &self.drift_intervals, + &mut state.drift_intervals, &scope.scope_digest, drift_kinds, observed_at, @@ -713,10 +874,10 @@ impl DaemonGitHubStackCoordinatorV1 { transitions, drift_observations, }; - self.observations - .lock() - .map_err(|_| GitHubStackCoordinatorErrorV1::Poisoned)? - .insert(scope.scope_digest, observation.clone()); + state.current = Some(PublishedObservation { + registration: origin.registration().cloned(), + observation: observation.clone(), + }); Ok(observation) } } diff --git a/crates/tracedecay-application/src/stack_coordinator/transition.rs b/crates/tracedecay-application/src/stack_coordinator/transition.rs index 032afbe366d..a5f52e2c2ca 100644 --- a/crates/tracedecay-application/src/stack_coordinator/transition.rs +++ b/crates/tracedecay-application/src/stack_coordinator/transition.rs @@ -1,6 +1,3 @@ -use std::collections::BTreeMap; -use std::sync::Mutex; - use serde::{Deserialize, Serialize}; use tracedecay_contracts::{ CancellationSignal, NativeIntegrationPreflightOutcomeV1, NativeIntegrationPreflightRequestV1, @@ -389,17 +386,16 @@ pub(super) fn drift_trace_id( } pub(super) fn update_drift_intervals( - intervals: &Mutex>>, + active: &mut Vec, scope_digest: &ManifestDigest, current_kinds: Vec, observed_at: UtcMicros, ) -> Result, GitHubStackCoordinatorErrorV1> { - let mut by_scope = intervals - .lock() - .map_err(|_| GitHubStackCoordinatorErrorV1::Poisoned)?; - let active = by_scope.entry(scope_digest.clone()).or_default(); + // Compute the complete next state before replacing the scope's intervals. + // An identity failure must not publish half of a transition. + let mut next = active.clone(); let mut observations = Vec::new(); - for interval in active + for interval in next .iter() .filter(|interval| !current_kinds.contains(&interval.kind)) { @@ -409,30 +405,30 @@ pub(super) fn update_drift_intervals( observed_at, )); } - active.retain(|interval| current_kinds.contains(&interval.kind)); + next.retain(|interval| current_kinds.contains(&interval.kind)); for kind in current_kinds { - let index = active.iter().position(|interval| interval.kind == kind); + let index = next.iter().position(|interval| interval.kind == kind); if index.is_none() { - active.push(DriftInterval { + next.push(DriftInterval { kind, trace_id: drift_trace_id(scope_digest, kind, observed_at)?, first_observed_at: observed_at, }); } observations.push(drift_observation( - &active[index.unwrap_or(active.len() - 1)], + &next[index.unwrap_or(next.len() - 1)], IntervalStateV1::Open, observed_at, )); } + *active = next; Ok(observations) } -pub(super) fn restore_open_drift_interval( - intervals: &Mutex>>, +pub(super) fn restored_open_interval( scope: &ResolvedScope, observation: &GitHubStackDriftObservationV1, -) -> Result<(), GitHubStackCoordinatorErrorV1> { +) -> Result { scope .validate() .map_err(|_| GitHubStackCoordinatorErrorV1::InvalidScope)?; @@ -440,33 +436,11 @@ pub(super) fn restore_open_drift_interval( if observation.drift.state != IntervalStateV1::Open { return Err(GitHubStackCoordinatorErrorV1::InvalidProviderObservation); } - let first_observed_at = UtcMicros(observation.drift.first_observed_micros); - if drift_trace_id( - &scope.scope_digest, - observation.drift.kind, - first_observed_at, - )? != observation.trace_id - { - return Err(GitHubStackCoordinatorErrorV1::InvalidProviderObservation); - } - let interval = DriftInterval { + Ok(DriftInterval { kind: observation.drift.kind, trace_id: observation.trace_id.clone(), - first_observed_at, - }; - let mut by_scope = intervals - .lock() - .map_err(|_| GitHubStackCoordinatorErrorV1::Poisoned)?; - let active = by_scope.entry(scope.scope_digest.clone()).or_default(); - if let Some(existing) = active - .iter_mut() - .find(|existing| existing.kind == interval.kind) - { - *existing = interval; - } else { - active.push(interval); - } - Ok(()) + first_observed_at: UtcMicros(observation.drift.first_observed_micros), + }) } fn push_unique(values: &mut Vec, value: T) { diff --git a/crates/tracedecay-application/tests/application_suite/github_stack_coordinator.rs b/crates/tracedecay-application/tests/application_suite/github_stack_coordinator.rs index 3777406ea31..077fc4849d4 100644 --- a/crates/tracedecay-application/tests/application_suite/github_stack_coordinator.rs +++ b/crates/tracedecay-application/tests/application_suite/github_stack_coordinator.rs @@ -4,7 +4,7 @@ mod preflight_tests; use std::collections::{BTreeMap, BTreeSet}; -use std::sync::Mutex; +use std::sync::{Arc, Barrier, Mutex}; use tracedecay_contracts::ResolvedScope; use tracedecay_domain::configuration::GitHubStackedPullRequestPolicyV1; @@ -163,6 +163,95 @@ fn provider_source_binding( } } +fn authorized( + coordinator: &DaemonGitHubStackCoordinatorV1, + exact_scope: &ResolvedScope, +) -> GitHubStackProviderReadV1 { + coordinator + .authorize_provider_read(exact_scope) + .unwrap() + .expect("a probing registration authorizes a provider read") +} + +const BASE_ADVANCE_TRANSITIONS: [StackSignalKindV1; 2] = [ + StackSignalKindV1::StackTipDrift, + StackSignalKindV1::PullRequestDrift, +]; + +/// The provider stack after its final target and bottom layer base advanced, +/// which opens exactly one base-advance drift interval against +/// [`enabled_snapshot`]. +fn advanced_snapshot(exact_scope: &ResolvedScope) -> GitHubStackProviderSnapshotV1 { + let mut advanced = enabled_snapshot(exact_scope); + advanced.response_digest = digest('f'); + advanced.final_target_commit_id = CommitId::new("commit.main.advanced").unwrap(); + advanced.layers[0].pull_request.base_commit_id = advanced.final_target_commit_id.clone(); + advanced +} + +/// Publishes the base stack and then its advance, returning the open +/// base-advance interval. +fn open_base_advance( + coordinator: &DaemonGitHubStackCoordinatorV1, + exact_scope: &ResolvedScope, +) -> GitHubStackDriftObservationV1 { + let provider = ProviderId::new("provider.github").unwrap(); + coordinator + .observe_provider( + authorized(coordinator, exact_scope), + provider.clone(), + GitHubStackProviderOutcomeV1::Enabled(enabled_snapshot(exact_scope)), + provider_source_binding(exact_scope, true), + UtcMicros(1), + ) + .unwrap(); + let opened = coordinator + .observe_provider( + authorized(coordinator, exact_scope), + provider, + GitHubStackProviderOutcomeV1::Enabled(advanced_snapshot(exact_scope)), + provider_source_binding(exact_scope, true), + UtcMicros(2), + ) + .unwrap(); + assert_eq!(opened.transitions, BASE_ADVANCE_TRANSITIONS); + let [open] = opened.drift_observations.as_slice() else { + panic!("an advanced base opens exactly one interval"); + }; + assert_eq!( + open.drift.kind, + tracedecay_domain::StackDriftKindV1::BaseAdvanced + ); + assert_eq!(open.drift.state, tracedecay_domain::IntervalStateV1::Open); + open.clone() +} + +/// The single drift observation must close `open` at `terminal`. +fn assert_closes( + exact_scope: &ResolvedScope, + observation: &GitHubStackObservationV1, + open: &GitHubStackDriftObservationV1, + terminal: i64, +) { + let [closed] = observation.drift_observations.as_slice() else { + panic!( + "expected exactly the open interval to close, got {:?}", + observation.drift_observations + ); + }; + assert_eq!(closed.trace_id, open.trace_id); + assert_eq!( + closed.drift.state, + tracedecay_domain::IntervalStateV1::Closed + ); + assert_eq!( + closed.drift.first_observed_micros, + open.drift.first_observed_micros + ); + assert_eq!(closed.drift.terminal_micros, Some(terminal)); + assert!(closed.validate(&exact_scope.scope_digest).is_ok()); +} + fn enabled_snapshot(exact_scope: &ResolvedScope) -> GitHubStackProviderSnapshotV1 { let source_anchor_id = RetrievalAnchorId::new("anchor.github.pull-request.41").unwrap(); GitHubStackProviderSnapshotV1 { @@ -232,7 +321,7 @@ fn production_coordinator_materializes_all_four_states_and_exact_stack_anchors() .unwrap(); let unavailable = coordinator .observe_provider( - probed.clone(), + authorized(&coordinator, &probed), provider.clone(), GitHubStackProviderOutcomeV1::Unavailable, provider_source_binding(&probed, false), @@ -245,7 +334,7 @@ fn production_coordinator_materializes_all_four_states_and_exact_stack_anchors() ); let degraded = coordinator .observe_provider( - probed.clone(), + authorized(&coordinator, &probed), provider.clone(), GitHubStackProviderOutcomeV1::Degraded { response_digest: digest('f'), @@ -260,7 +349,7 @@ fn production_coordinator_materializes_all_four_states_and_exact_stack_anchors() ); let enabled = coordinator .observe_provider( - probed.clone(), + authorized(&coordinator, &probed), provider, GitHubStackProviderOutcomeV1::Enabled(enabled_snapshot(&probed)), provider_source_binding(&probed, true), @@ -295,14 +384,53 @@ fn production_coordinator_materializes_all_four_states_and_exact_stack_anchors() assert_eq!( coordinator .observe_policy( - probed, + probed.clone(), ProviderId::new("provider.github").unwrap(), - provider_source_binding(&scope(1), false), + provider_source_binding(&probed, false), UtcMicros(5), ) .unwrap(), enabled ); + coordinator + .register_scope(&probed, GitHubStackedPullRequestPolicyV1::Disabled) + .unwrap(); + coordinator + .register_scope( + &probed, + GitHubStackedPullRequestPolicyV1::ProbePrivatePreview, + ) + .unwrap(); + let reenabled = coordinator + .observe_policy( + probed.clone(), + ProviderId::new("provider.github").unwrap(), + provider_source_binding(&probed, false), + UtcMicros(6), + ) + .unwrap(); + assert_eq!( + reenabled.capability.state, + tracedecay_domain::GitHubStackCapabilityStateV1::Unavailable, + "reenabling the provider requires a fresh observation" + ); + assert!(reenabled.snapshot.is_none()); + coordinator + .register_scope(&probed, GitHubStackedPullRequestPolicyV1::Disabled) + .unwrap(); + let disabled = coordinator + .observe_policy( + probed.clone(), + ProviderId::new("provider.github").unwrap(), + provider_source_binding(&probed, false), + UtcMicros(7), + ) + .unwrap(); + assert_eq!( + disabled.capability.state, + tracedecay_domain::GitHubStackCapabilityStateV1::PrivatePreviewDisabled + ); + assert!(disabled.snapshot.is_none()); } #[test] @@ -318,7 +446,7 @@ fn mounted_provider_observation_reports_pr_tip_merge_base_and_ci_drift() { .unwrap(); coordinator .observe_provider( - exact_scope.clone(), + authorized(&coordinator, &exact_scope), provider.clone(), GitHubStackProviderOutcomeV1::Enabled(enabled_snapshot(&exact_scope)), provider_source_binding(&exact_scope, true), @@ -336,7 +464,7 @@ fn mounted_provider_observation_reports_pr_tip_merge_base_and_ci_drift() { changed.layers[0].ci_digest = digest('a'); let observation = coordinator .observe_provider( - exact_scope.clone(), + authorized(&coordinator, &exact_scope), provider.clone(), GitHubStackProviderOutcomeV1::Enabled(changed.clone()), provider_source_binding(&exact_scope, true), @@ -390,13 +518,50 @@ fn mounted_provider_observation_reports_pr_tip_merge_base_and_ci_drift() { ), Err(GitHubStackCoordinatorErrorV1::InvalidProviderObservation) ); - DaemonGitHubStackCoordinatorV1::default() + let restored = DaemonGitHubStackCoordinatorV1::default(); + restored .restore_open_drift_interval(&exact_scope, &persisted) .unwrap(); + assert_eq!( + restored.observe_policy( + exact_scope.clone(), + provider.clone(), + provider_source_binding(&exact_scope, false), + UtcMicros(1), + ), + Err(GitHubStackCoordinatorErrorV1::InvalidProviderObservation) + ); + let settled = restored + .observe_policy( + exact_scope.clone(), + provider.clone(), + provider_source_binding(&exact_scope, false), + UtcMicros(3), + ) + .unwrap(); + assert_eq!(settled.drift_observations.len(), 1); + assert_eq!(settled.drift_observations[0].trace_id, persisted.trace_id); + assert!( + settled.drift_observations[0] + .validate(&exact_scope.scope_digest) + .is_ok() + ); + // A provider completion can reach the transition lock after a newer sample. + // It cannot close the open intervals before their recorded starting time. + assert_eq!( + coordinator.observe_provider( + authorized(&coordinator, &exact_scope), + provider.clone(), + GitHubStackProviderOutcomeV1::Enabled(changed.clone()), + provider_source_binding(&exact_scope, true), + UtcMicros(1), + ), + Err(GitHubStackCoordinatorErrorV1::InvalidProviderObservation) + ); let closed = coordinator .observe_provider( - exact_scope.clone(), + authorized(&coordinator, &exact_scope), provider, GitHubStackProviderOutcomeV1::Enabled(changed), provider_source_binding(&exact_scope, true), @@ -411,6 +576,270 @@ fn mounted_provider_observation_reports_pr_tip_merge_base_and_ci_drift() { })); } +#[test] +fn concurrent_identical_provider_publications_have_one_drift_transition() { + let coordinator = Arc::new(DaemonGitHubStackCoordinatorV1::default()); + let exact_scope = scope(94); + let provider = ProviderId::new("provider.github").unwrap(); + coordinator + .register_scope( + &exact_scope, + GitHubStackedPullRequestPolicyV1::ProbePrivatePreview, + ) + .unwrap(); + coordinator + .observe_provider( + authorized(&coordinator, &exact_scope), + provider.clone(), + GitHubStackProviderOutcomeV1::Enabled(enabled_snapshot(&exact_scope)), + provider_source_binding(&exact_scope, true), + UtcMicros(1), + ) + .unwrap(); + + let mut changed = enabled_snapshot(&exact_scope); + changed.response_digest = digest('f'); + changed.final_target_commit_id = CommitId::new("commit.main.changed").unwrap(); + changed.layers[0].pull_request.base_commit_id = changed.final_target_commit_id.clone(); + let barrier = Arc::new(Barrier::new(24)); + let observations = std::thread::scope(|threads| { + let joins = (0..24) + .map(|_| { + let coordinator = Arc::clone(&coordinator); + let exact_scope = exact_scope.clone(); + let provider = provider.clone(); + let changed = changed.clone(); + let barrier = Arc::clone(&barrier); + threads.spawn(move || { + let read = authorized(&coordinator, &exact_scope); + barrier.wait(); + coordinator + .observe_provider( + read, + provider, + GitHubStackProviderOutcomeV1::Enabled(changed), + provider_source_binding(&exact_scope, true), + UtcMicros(2), + ) + .unwrap() + }) + }) + .collect::>(); + joins + .into_iter() + .map(|join| join.join().unwrap()) + .collect::>() + }); + assert_eq!( + observations + .iter() + .filter(|observation| { + observation + .transitions + .contains(&StackSignalKindV1::StackTipDrift) + }) + .count(), + 1, + "one changed head must produce one transition, even with concurrent publishers" + ); + assert_eq!( + observations + .iter() + .flat_map(|observation| &observation.drift_observations) + .filter(|event| { + event.drift.kind == tracedecay_domain::StackDriftKindV1::BaseAdvanced + && event.drift.state == tracedecay_domain::IntervalStateV1::Open + }) + .count(), + 1, + "only the winning transition may open a base-advance interval" + ); + assert!(observations.iter().all(|observation| { + observation + .drift_observations + .iter() + .all(|event| event.validate(&exact_scope.scope_digest).is_ok()) + })); +} + +#[test] +fn provider_read_spanning_disable_and_reenable_cannot_publish() { + let coordinator = DaemonGitHubStackCoordinatorV1::default(); + let exact_scope = scope(96); + let provider = ProviderId::new("provider.github").unwrap(); + let probe = GitHubStackedPullRequestPolicyV1::ProbePrivatePreview; + coordinator.register_scope(&exact_scope, probe).unwrap(); + let open = open_base_advance(&coordinator, &exact_scope); + + // Disabling and re-enabling while the read runs leaves the policy equal + // again; the completion still belongs to the retired registration. + let held = authorized(&coordinator, &exact_scope); + coordinator + .register_scope(&exact_scope, GitHubStackedPullRequestPolicyV1::Disabled) + .unwrap(); + assert!( + coordinator + .authorize_provider_read(&exact_scope) + .unwrap() + .is_none(), + "a disabled registration authorizes no provider read" + ); + coordinator.register_scope(&exact_scope, probe).unwrap(); + assert_eq!( + coordinator.observe_provider( + held, + provider.clone(), + GitHubStackProviderOutcomeV1::Enabled(enabled_snapshot(&exact_scope)), + provider_source_binding(&exact_scope, true), + UtcMicros(5), + ), + Err(GitHubStackCoordinatorErrorV1::Superseded) + ); + + // The refusal moved neither the snapshot nor the drift intervals. A new + // read finishing before the refused sample's timestamp still passes the + // fence, and closes exactly the interval the retired registration opened. + let fresh = coordinator + .observe_provider( + authorized(&coordinator, &exact_scope), + provider, + GitHubStackProviderOutcomeV1::Enabled(advanced_snapshot(&exact_scope)), + provider_source_binding(&exact_scope, true), + UtcMicros(3), + ) + .unwrap(); + assert_eq!( + fresh.capability.state, + tracedecay_domain::GitHubStackCapabilityStateV1::Enabled + ); + assert!(fresh.snapshot.is_some()); + assert!( + fresh.transitions.is_empty(), + "a retired registration's snapshot is not a drift baseline" + ); + assert_closes(&exact_scope, &fresh, &open, 3); +} + +#[test] +fn abandoned_unchanged_and_disabled_registrations_keep_scope_state_exact() { + let coordinator = DaemonGitHubStackCoordinatorV1::default(); + let exact_scope = scope(97); + let provider = ProviderId::new("provider.github").unwrap(); + let probe = GitHubStackedPullRequestPolicyV1::ProbePrivatePreview; + assert!( + coordinator + .authorize_provider_read(&exact_scope) + .unwrap() + .is_none(), + "an unregistered scope authorizes no provider read" + ); + coordinator.register_scope(&exact_scope, probe).unwrap(); + coordinator + .observe_provider( + authorized(&coordinator, &exact_scope), + provider.clone(), + GitHubStackProviderOutcomeV1::Enabled(enabled_snapshot(&exact_scope)), + provider_source_binding(&exact_scope, true), + UtcMicros(1), + ) + .unwrap(); + + // A cancelled refresh drops its read; nothing is left to clean up. + drop(authorized(&coordinator, &exact_scope)); + // Re-registering the unchanged policy, as each project open does, keeps + // the registration, its in-flight read and its drift baseline. + let in_flight = authorized(&coordinator, &exact_scope); + coordinator.register_scope(&exact_scope, probe).unwrap(); + let drifted = coordinator + .observe_provider( + in_flight, + provider.clone(), + GitHubStackProviderOutcomeV1::Enabled(advanced_snapshot(&exact_scope)), + provider_source_binding(&exact_scope, true), + UtcMicros(2), + ) + .unwrap(); + assert_eq!(drifted.transitions, BASE_ADVANCE_TRANSITIONS); + let [open] = drifted.drift_observations.as_slice() else { + panic!("an advanced base opens exactly one interval"); + }; + + // A read spanning a plain disable is refused, not republished as the + // disabled policy. The policy observation then closes the open interval. + let spanning = authorized(&coordinator, &exact_scope); + coordinator + .register_scope(&exact_scope, GitHubStackedPullRequestPolicyV1::Disabled) + .unwrap(); + assert_eq!( + coordinator.observe_provider( + spanning, + provider.clone(), + GitHubStackProviderOutcomeV1::Unavailable, + provider_source_binding(&exact_scope, false), + UtcMicros(3), + ), + Err(GitHubStackCoordinatorErrorV1::Superseded) + ); + let disabled = coordinator + .observe_policy( + exact_scope.clone(), + provider, + provider_source_binding(&exact_scope, false), + UtcMicros(3), + ) + .unwrap(); + assert_eq!( + disabled.capability.state, + tracedecay_domain::GitHubStackCapabilityStateV1::PrivatePreviewDisabled + ); + assert_closes(&exact_scope, &disabled, open, 3); +} + +#[test] +fn restarted_owner_refuses_pre_restart_reads_and_settles_restored_drift() { + let exact_scope = scope(98); + let provider = ProviderId::new("provider.github").unwrap(); + let probe = GitHubStackedPullRequestPolicyV1::ProbePrivatePreview; + let before = DaemonGitHubStackCoordinatorV1::default(); + before.register_scope(&exact_scope, probe).unwrap(); + let open = open_base_advance(&before, &exact_scope); + let persisted = GitHubStackDriftObservationV1::from_persisted( + &exact_scope.scope_digest, + open.observed_at, + open.drift.clone(), + ) + .unwrap(); + let pre_restart = authorized(&before, &exact_scope); + + // The same scope and policy registered again by a restarted owner is a + // different registration. + let restarted = DaemonGitHubStackCoordinatorV1::default(); + restarted + .restore_open_drift_interval(&exact_scope, &persisted) + .unwrap(); + restarted.register_scope(&exact_scope, probe).unwrap(); + assert_eq!( + restarted.observe_provider( + pre_restart, + provider.clone(), + GitHubStackProviderOutcomeV1::Enabled(enabled_snapshot(&exact_scope)), + provider_source_binding(&exact_scope, true), + UtcMicros(4), + ), + Err(GitHubStackCoordinatorErrorV1::Superseded) + ); + let settled = restarted + .observe_provider( + authorized(&restarted, &exact_scope), + provider, + GitHubStackProviderOutcomeV1::Enabled(advanced_snapshot(&exact_scope)), + provider_source_binding(&exact_scope, true), + UtcMicros(3), + ) + .unwrap(); + assert_closes(&exact_scope, &settled, &persisted, 3); +} + #[tokio::test] async fn mounted_owner_dedupes_debounced_signals_and_appends_all_material_overflow() { let coordinator = DaemonGitHubStackCoordinatorV1::default(); @@ -687,7 +1116,7 @@ fn daemon_restart_returns_unavailable_until_a_delayed_exact_probe_upgrades_state let upgraded = restarted .observe_provider( - exact_scope, + authorized(&restarted, &exact_scope), provider, GitHubStackProviderOutcomeV1::Enabled(enabled_snapshot(&scope(80))), provider_source_binding(&scope(80), true),