diff --git a/crates/registry-scheduling-client/src/client.rs b/crates/registry-scheduling-client/src/client.rs index b131ddbcf..0684a9562 100644 --- a/crates/registry-scheduling-client/src/client.rs +++ b/crates/registry-scheduling-client/src/client.rs @@ -11,13 +11,14 @@ use registry_platform_httputil::{ read_bounded, url::append_path_segments, validate_response_headers, }; use registry_scheduling_core::{ - type_uri, AdmissionRequest, AppointmentDocument, AppointmentHistoryEntryDocument, - AvailabilityEntry, CancelAppointmentRequest, CreateAppointmentRequest, ExplainDocument, - HoldDocument, LocationDocument, OfferingDocument, PageDocument, ProblemCode, - RescheduleAppointmentRequest, ResourceDocument, SchedulingServiceDocument, ServiceDocument, - APPOINTMENTS_PATH, AVAILABILITY_EXPLAIN_PATH, AVAILABILITY_PATH, CURSOR_QUERY_PARAMETER, - HOLDS_PATH, IDEMPOTENCY_KEY_HEADER, LIMIT_QUERY_PARAMETER, LOCATIONS_PATH, - MAXIMUM_IDEMPOTENCY_KEY_BYTES, OFFERINGS_PATH, RESOURCES_PATH, SCHEDULING_PATH, SERVICES_PATH, + type_uri, valid_identifier, AdmissionRequest, AppointmentDocument, + AppointmentHistoryEntryDocument, AvailabilityEntry, CancelAppointmentRequest, + CreateAppointmentRequest, ExplainDocument, HoldDocument, LocationDocument, OfferingDocument, + PageDocument, ProblemCode, RescheduleAppointmentRequest, ResourceDocument, + SchedulingServiceDocument, ServiceDocument, APPOINTMENTS_PATH, AVAILABILITY_EXPLAIN_PATH, + AVAILABILITY_PATH, CURSOR_QUERY_PARAMETER, HOLDS_PATH, IDEMPOTENCY_KEY_HEADER, + LIMIT_QUERY_PARAMETER, LOCATIONS_PATH, MAXIMUM_IDEMPOTENCY_KEY_BYTES, OFFERINGS_PATH, + RESOURCES_PATH, SCHEDULING_PATH, SERVICES_PATH, }; use reqwest::header::{HeaderName, HeaderValue, ACCEPT, AUTHORIZATION, CONTENT_TYPE}; use reqwest::{RequestBuilder, Response, StatusCode, Url}; @@ -142,11 +143,7 @@ impl SchedulingClient { offering: &str, start: DateTime, ) -> Result, SchedulingClientError> { - if offering.is_empty() { - return Err(SchedulingClientError::invalid_request( - "the offering selector is invalid", - )); - } + validate_offering(offering)?; let query = [ (OFFERING_QUERY_PARAMETER, offering.to_owned()), (START_QUERY_PARAMETER, rfc3339(start)), @@ -564,6 +561,22 @@ fn validate_identifier( Ok(()) } +/// The offering selector both availability routes carry. It travels as a +/// query parameter rather than a path segment, so an out-of-grammar value +/// cannot reach a different route; what it would otherwise buy is a round +/// trip spent sending a URL the runtime's own grammar already refuses. An +/// offering is authored in a policy rather than minted into a document, so +/// the grammar it is held to here is the one the runtime publishes offerings +/// under, not the wider alphabet an opaque route identifier travels in. +fn validate_offering(offering: &str) -> Result<(), SchedulingClientError> { + if valid_identifier(offering) { + return Ok(()); + } + Err(SchedulingClientError::invalid_request( + "the offering selector is invalid", + )) +} + fn validate_availability( offering: &str, _start: Option>, @@ -571,11 +584,7 @@ fn validate_availability( cursor: Option<&str>, limit: Option, ) -> Result<(), SchedulingClientError> { - if offering.is_empty() { - return Err(SchedulingClientError::invalid_request( - "the offering selector is invalid", - )); - } + validate_offering(offering)?; validate_cursor(cursor)?; if limit.is_some_and(|value| value == 0) { return Err(SchedulingClientError::invalid_request( @@ -648,6 +657,38 @@ mod tests { assert!(validate_availability("registry-update-30", None, None, None, Some(0)).is_err()); } + /// Both availability entry points refuse an out-of-grammar selector + /// before any I/O, against the grammar the runtime publishes offerings + /// under rather than the looser one a route segment carries: a selector + /// the runtime would refuse on arrival is refused here instead. + #[test] + fn offering_selectors_are_bounded_before_io() { + assert!(validate_offering("registry-update-30").is_ok()); + for foreign in [ + "", + "registry/update", + "registry update", + "registry?update", + // Each of these is a legal opaque route segment and no + // identifier the policy grammar admits. + "Registry_Update", + "registry_update", + "registry.update", + "30-minute-update", + &"x".repeat(65), + &"x".repeat(MAXIMUM_IDENTIFIER_BYTES + 1), + ] { + assert!( + validate_offering(foreign).is_err(), + "{foreign:?} must be refused" + ); + assert!( + validate_availability(foreign, None, None, None, None).is_err(), + "{foreign:?} must be refused by availability" + ); + } + } + #[test] fn availability_ranges_preserve_runtime_normalization() { let start = Utc.with_ymd_and_hms(2026, 10, 5, 2, 0, 0).unwrap(); diff --git a/crates/registry-scheduling-core/src/diagnostics.rs b/crates/registry-scheduling-core/src/diagnostics.rs index 629ee9926..6eef2cb65 100644 --- a/crates/registry-scheduling-core/src/diagnostics.rs +++ b/crates/registry-scheduling-core/src/diagnostics.rs @@ -56,6 +56,10 @@ pub enum PolicyCheckReason { /// Two declarations draw on the same supply without a partition the /// ledger can enforce. SharedSupplyUnpartitioned, + /// A resource pool and a published window claim the same identifier. + /// Both anchor their capacity transactions on one row keyed by that + /// identifier, so the two kinds of supply must not share one. + SupplyIdentifierCollision, /// A local hook omits or changes the shared handler ABI. UnsupportedHookAbi, /// A hook phase and handler kind cannot run together. @@ -101,6 +105,7 @@ impl PolicyCheckReason { Self::InvalidBands => "invalid-bands", Self::SubquotaOverdrawn => "subquota-overdrawn", Self::SharedSupplyUnpartitioned => "shared-supply-unpartitioned", + Self::SupplyIdentifierCollision => "supply-identifier-collision", Self::UnsupportedHookAbi => "unsupported-hook-abi", Self::UnsupportedHookPhase => "unsupported-hook-phase", Self::UnsupportedHookTrigger => "unsupported-hook-trigger", diff --git a/crates/registry-scheduling-core/src/model.rs b/crates/registry-scheduling-core/src/model.rs index 4b3b07a49..32ab49cbf 100644 --- a/crates/registry-scheduling-core/src/model.rs +++ b/crates/registry-scheduling-core/src/model.rs @@ -158,6 +158,11 @@ impl LedgerSnapshot { /// Whether a consuming claim already holds `duplicate_key` at `now`, /// ignoring `exclude`. + /// + /// The key admits one live claim per party for an offering, so a claim + /// holds it only until the time it occupies has passed. A booking keeps + /// consuming capacity for overlap forever, but a party whose appointment + /// is behind them is free to book the offering again. pub fn duplicate_active( &self, offering: &str, @@ -169,6 +174,7 @@ impl LedgerSnapshot { claim.offering == offering && claim.duplicate_key.as_deref() == Some(duplicate_key) && Some(claim.id.as_str()) != exclude + && claim.end > now }) } } @@ -574,6 +580,7 @@ mod tests { id: "claim-2".to_owned(), offering: "offering-b".to_owned(), kind: LedgerKind::Booking, + end: utc(5, 0), expires_at: None, ..expired_hold.clone() }; @@ -584,6 +591,30 @@ mod tests { assert!(snapshot.duplicate_active("offering-b", "subject:one", None, now)); } + #[test] + fn an_elapsed_booking_stops_holding_the_duplicate_key() { + let booking = LedgerClaim { + id: "claim-1".to_owned(), + offering: "offering-a".to_owned(), + supply_id: "morning-window".to_owned(), + kind: LedgerKind::Booking, + channel: None, + start: utc(1, 0), + end: utc(2, 0), + units: 1, + duplicate_key: Some("subject:one".to_owned()), + expires_at: None, + }; + let snapshot = LedgerSnapshot { + claims: vec![booking], + }; + // The key names one live booking per party, so it is held while the + // appointment stands and released once the appointment has passed. + assert!(snapshot.duplicate_active("offering-a", "subject:one", None, utc(1, 30))); + assert!(!snapshot.duplicate_active("offering-a", "subject:one", None, utc(2, 0))); + assert!(!snapshot.duplicate_active("offering-a", "subject:one", None, utc(3, 0))); + } + #[test] fn the_same_request_always_hashes_identically() { assert_eq!( diff --git a/crates/registry-scheduling-core/src/policy.rs b/crates/registry-scheduling-core/src/policy.rs index b70b74521..94def9e50 100644 --- a/crates/registry-scheduling-core/src/policy.rs +++ b/crates/registry-scheduling-core/src/policy.rs @@ -544,6 +544,16 @@ impl SchedulingPolicy { for (index, offering) in self.offerings.iter().enumerate() { self.check_offering(offering, index, &mut findings); } + for (index, offering) in self.offerings.iter().enumerate() { + if let Some(arrival) = &offering.arrival { + if self.sells_pool(&arrival.window) { + findings.push(SchedulingDiagnostic::new( + format!("offerings[{index}].arrival.window"), + PolicyCheckReason::SupplyIdentifierCollision, + )); + } + } + } check_because( &self.hold_policy.because, @@ -691,6 +701,21 @@ impl SchedulingPolicy { findings, ); } + // A request carries the capabilities and prerequisites that satisfy + // these, and the request edge bounds its own lists at the same + // maximum. An offering requiring more than that bound publishes + // cleanly and refuses every admissible request, so it is refused + // here instead. + check_collection_bound( + &offering.requires_capabilities, + &format!("{path}.requiresCapabilities"), + findings, + ); + check_collection_bound( + &offering.prerequisites, + &format!("{path}.prerequisites"), + findings, + ); for capability in &offering.requires_capabilities { if !valid_identifier(capability) { findings.push(SchedulingDiagnostic::new( @@ -778,6 +803,19 @@ impl SchedulingPolicy { } findings.extend(window.units_policy.check(&format!("{path}.unitsPolicy"))); + // A pool and a window anchor their capacity transactions on one row + // keyed by the identifier, so the two supply namespaces must stay + // disjoint. Sharing an identifier makes the two publish paths write + // the same row: one aborts on the key, the other leaves the wrong + // kind standing and the next records swap deletes the anchor the + // pool still depends on. + if self.sells_pool(&window.id) { + findings.push(SchedulingDiagnostic::new( + format!("{path}.id"), + PolicyCheckReason::SupplyIdentifierCollision, + )); + } + if window.leftover.is_some() { findings.push(SchedulingDiagnostic::new( format!("{path}.leftover"), @@ -893,6 +931,18 @@ impl SchedulingPolicy { } } + /// Whether any exact-time offering draws on the pool named `id`. Those + /// are the pools publication anchors, so they are the identifiers a + /// window may not take. + fn sells_pool(&self, id: &str) -> bool { + self.offerings.iter().any(|offering| { + offering + .exact_time + .as_ref() + .is_some_and(|exact| exact.pool == id) + }) + } + fn check_unique_id( &self, id: &str, @@ -1715,6 +1765,69 @@ windows: ); } + /// A resource pool and a published window anchor their capacity + /// transactions on one row keyed by the identifier, so the two supply + /// namespaces have to stay disjoint. The policy states one side of the + /// collision by itself, and the records state the other, so each + /// authored document is refused naming its own field. + #[test] + fn a_pool_and_a_window_may_not_share_one_supply_identifier() { + let mut policy = household_window_policy(); + let windows = household_windows(); + assert!(policy.check().is_empty(), "{:?}", policy.check()); + assert!( + policy.check_window_records(&windows).is_empty(), + "{:?}", + policy.check_window_records(&windows) + ); + + policy.offerings.push(OfferingPolicy { + id: "urgent-five".to_owned(), + service: "household-day".to_owned(), + label: "Urgent five-minute slot".to_owned(), + mode: SchedulingMode::ExactTime, + location: "civic-hall".to_owned(), + because: "Urgent matters get exact five-minute slots.".to_owned(), + exact_time: Some(ExactTimeOffering { + duration_minutes: 5, + buffer_before_minutes: 0, + buffer_after_minutes: 0, + lead_time_minutes: 30, + horizon_days: 14, + pool: "household-morning-window".to_owned(), + start_increment_minutes: 5, + max_recipients: 1, + }), + arrival: None, + cancellation_cutoff_minutes: 60, + reminders: Vec::new(), + duplicate_active_key: None, + requires_capabilities: Vec::new(), + prerequisites: Vec::new(), + }); + + // The policy alone already names both sides: the arrival offering + // points at a window identifier its own exact-time offering sells. + let rendered: Vec = policy.check().iter().map(ToString::to_string).collect(); + assert!( + rendered + .contains(&"offerings[0].arrival.window: supply-identifier-collision".to_owned()), + "{rendered:?}" + ); + + // The published record is refused in its own vocabulary, which is + // the finding the two database writes carry. + let rendered: Vec = policy + .check_window_records(&windows) + .iter() + .map(ToString::to_string) + .collect(); + assert!( + rendered.contains(&"windows[0].id: supply-identifier-collision".to_owned()), + "{rendered:?}" + ); + } + /// AT-22, static half: an unpartitioned staffing claim over a pool that an /// exact-time offering also sells is rejected at publication. /// @@ -2435,4 +2548,31 @@ surprise: true let at_bound = minimal_exact_time_policy(); assert!(at_bound.check().is_empty()); } + + /// An offering may require no more capabilities or prerequisites than a + /// request is allowed to carry. Past that bound the offering publishes + /// cleanly and no admissible request can ever satisfy it, so the refusal + /// belongs at authoring time where the operator can still name it. + #[test] + fn offering_requirement_collections_are_bounded() { + let mut policy = minimal_exact_time_policy(); + policy.offerings[0].requires_capabilities = (0..=MAXIMUM_COLLECTION_ENTRIES) + .map(|index| format!("capability-{index}")) + .collect(); + policy.offerings[0].prerequisites = (0..=MAXIMUM_COLLECTION_ENTRIES) + .map(|index| format!("urn:evidence:residency-{index}")) + .collect(); + let rendered: Vec = policy.check().iter().map(|f| f.to_string()).collect(); + assert!(rendered.contains(&"offerings[0].requiresCapabilities: invalid-bound".to_owned())); + assert!(rendered.contains(&"offerings[0].prerequisites: invalid-bound".to_owned())); + + let mut at_bound = minimal_exact_time_policy(); + at_bound.offerings[0].requires_capabilities = (1..=MAXIMUM_COLLECTION_ENTRIES) + .map(|index| format!("capability-{index}")) + .collect(); + at_bound.offerings[0].prerequisites = (1..=MAXIMUM_COLLECTION_ENTRIES) + .map(|index| format!("urn:evidence:residency-{index}")) + .collect(); + assert!(at_bound.check().is_empty()); + } } diff --git a/crates/registry-scheduling-core/src/problem.rs b/crates/registry-scheduling-core/src/problem.rs index 1b05e820f..f8fc714e9 100644 --- a/crates/registry-scheduling-core/src/problem.rs +++ b/crates/registry-scheduling-core/src/problem.rs @@ -266,7 +266,7 @@ impl ProblemCode { "The bearer credential is missing, invalid, or expired. Sign in again." } Self::BookingDuplicateActive => { - "An active booking already holds this party's duplicate key. Cancel or complete it before booking again." + "An active booking already holds this party's duplicate key. Cancel it, or wait until it has passed, before booking again." } Self::CancellationCutoffPassed => { "The cancellation cutoff for this appointment has passed, so it can no longer be cancelled." diff --git a/crates/registry-scheduling/migrations/0007_duplicate_lookup_index.sql b/crates/registry-scheduling/migrations/0007_duplicate_lookup_index.sql new file mode 100644 index 000000000..af9bab19e --- /dev/null +++ b/crates/registry-scheduling/migrations/0007_duplicate_lookup_index.sql @@ -0,0 +1,13 @@ +-- The offering-wide duplicate guard filters on three columns: the key the +-- caller sent, the offering the claim was made under, and whether the booking +-- it names is still ahead. Schema version 1 indexed the key alone, leaving the +-- other two to a recheck over every active claim carrying that key. A booking +-- stays 'active' once its time has passed, because the ledger publishes no +-- completion transition and nothing sweeps elapsed bookings closed, so the +-- rechecked set grows for the life of the deployment rather than with the +-- bookings a party currently holds. Index the whole predicate instead. +CREATE INDEX IF NOT EXISTS scheduling_claims_duplicate_lookup_idx + ON scheduling_claims(duplicate_key, offering, occupied_end) + WHERE state = 'active' AND duplicate_key IS NOT NULL; + +DROP INDEX IF EXISTS scheduling_claims_duplicate_idx; diff --git a/crates/registry-scheduling/src/service.rs b/crates/registry-scheduling/src/service.rs index b45683051..bc95f2fbf 100644 --- a/crates/registry-scheduling/src/service.rs +++ b/crates/registry-scheduling/src/service.rs @@ -31,6 +31,7 @@ use registry_scheduling_core::{ }; use serde_json::{json, Value}; use sha2::{Digest as _, Sha256}; +use std::borrow::Cow; use uuid::Uuid; use crate::cursors::{ @@ -624,11 +625,9 @@ impl SchedulingService { if hold.kind != LedgerKind::Hold { return Err(ServiceError::Problem(ProblemCode::HoldReleased)); } - let offering = self.policy.offering(&hold.offering).ok_or_else(|| { - ServiceError::internal("a committed hold names no offering in the policy") - })?; + let offering = self.claim_offering(&hold, now).await?; let grant = self - .require_permission(caller, offering, HOLD_RELEASE_ACTION) + .require_permission(caller, &offering, HOLD_RELEASE_ACTION) .await?; let actor = caller.actor_pseudonym(&self.hasher, &self.scheduling_id)?; let request_hash = canonical_hash(&json!({"hold": hold_id}))?; @@ -891,11 +890,9 @@ impl SchedulingService { .booking(appointment_id) .await? .ok_or(ServiceError::Problem(ProblemCode::OperationNotAuthorized))?; - let offering = self.policy.offering(&appointment.offering).ok_or_else(|| { - ServiceError::internal("a committed appointment names no offering in the policy") - })?; + let offering = self.claim_offering(&appointment, now).await?; let grant = self - .require_permission(caller, offering, APPOINTMENT_CANCEL_ACTION) + .require_permission(caller, &offering, APPOINTMENT_CANCEL_ACTION) .await?; let actor = caller.actor_pseudonym(&self.hasher, &self.scheduling_id)?; let request_hash = canonical_hash(&json!({ @@ -1180,6 +1177,42 @@ impl SchedulingService { } } + /// The offering a committed claim names. + /// + /// The current policy answers for a claim that still pins it, because + /// publication refuses to change the service, location, mode or supply of + /// an offering while such a claim stands: the live terms and the ones the + /// claim was committed under are the same terms. Every other claim, a + /// closed one or a hold already past its expiry, pins nothing. Its + /// offering may since have been retired, or kept its identifier and been + /// sold under a different service, and the retry its receipt exists to + /// serve still has to be authorized before that receipt is replayed. So + /// the terms are read from the policy revision the claim itself names, + /// and the grant is matched against the offering as it stood when the + /// claim was committed: the caller who committed it can still close it, + /// and a grant over whatever the identifier sells today cannot reach a + /// receipt written under something else. + async fn claim_offering( + &self, + claim: &ClaimRow, + now: DateTime, + ) -> Result, ServiceError> { + if claim.pins_its_offering(now) { + if let Some(current) = self.policy.offering(&claim.offering) { + return Ok(Cow::Borrowed(current)); + } + } + self.store + .retained_offering(claim.policy_revision, &claim.offering) + .await? + .map(Cow::Owned) + .ok_or_else(|| { + ServiceError::internal( + "a committed claim names no offering in the policy revision it was committed under", + ) + }) + } + /// The policy-resolved supply an offering runs against, read from the /// live facts the operator's records apply wrote, with the records /// revision the read observed. The revision travels into the capacity diff --git a/crates/registry-scheduling/src/store.rs b/crates/registry-scheduling/src/store.rs index 13dd55297..77127f673 100644 --- a/crates/registry-scheduling/src/store.rs +++ b/crates/registry-scheduling/src/store.rs @@ -43,7 +43,8 @@ use registry_platform_config::SecretResolver; use registry_scheduling_core::{ assess_window_record_impact, evaluate_exact_time_admission, evaluate_hold_state, evaluate_window_admission, AdmissionRefusal, ExactTimeContext, LedgerClaim, LedgerKind, - LedgerSnapshot, PoolMember, SchedulingFacts, SchedulingPolicy, APPOINTMENT_CANCELLED_TRIGGER, + LedgerSnapshot, PolicyCheckReason, PoolMember, PublishedWindow, SchedulingDiagnostic, + SchedulingFacts, SchedulingPolicy, APPOINTMENT_CANCELLED_TRIGGER, APPOINTMENT_CONFIRMED_TRIGGER, APPOINTMENT_RESCHEDULED_TRIGGER, }; use serde::{Deserialize, Serialize}; @@ -66,16 +67,20 @@ const WINDOW_RECORDS_MIGRATION_VERSION: i64 = 5; const WINDOW_REVISION_HEADS_MIGRATION: &str = include_str!("../migrations/0006_window_revision_heads.sql"); const WINDOW_REVISION_HEADS_MIGRATION_VERSION: i64 = 6; +const DUPLICATE_LOOKUP_INDEX_MIGRATION: &str = + include_str!("../migrations/0007_duplicate_lookup_index.sql"); +const DUPLICATE_LOOKUP_INDEX_MIGRATION_VERSION: i64 = 7; /// Every schema version in ledger order. const MIGRATIONS: [(i64, &str); 2] = [(1, SCHEDULING_MIGRATION), (2, FACTS_REVISION_MIGRATION)]; -const SCHEMA_VERSIONS: [i64; 6] = [ +const SCHEMA_VERSIONS: [i64; 7] = [ 1, 2, HOOK_DELIVERY_MIGRATION_VERSION, POLICY_DOCUMENT_MIGRATION_VERSION, WINDOW_RECORDS_MIGRATION_VERSION, WINDOW_REVISION_HEADS_MIGRATION_VERSION, + DUPLICATE_LOOKUP_INDEX_MIGRATION_VERSION, ]; /// Serializes operator-run migrations on one session lock. A second migrator @@ -133,6 +138,25 @@ pub enum StoreError { }, #[error("the proposed policy would strand standing commitments: {0}")] PolicyInUse(String), + /// A commitment asked to serialize against supply no publication + /// anchored. Naming the reference is what makes this an operator's + /// deployment gap in the diagnostics rather than an unexplained + /// inconsistency; the caller is told only that the deployment is + /// unavailable. + #[error("the Scheduling deployment anchors no supply for {0}")] + UnanchoredSupply(String), + /// A resource pool and a published window claim the same supply + /// identifier. The two publish paths would write one anchor row between + /// them, so the identifier is refused by name at whichever of them + /// arrives second. + #[error("the supply identifier {0}")] + SupplyIdentifierCollision(String), + /// A policy and the window records it governs contradict each other. + /// The authoring tool refuses the same combination against the files on + /// an operator's disk; this is that refusal taken at the write, where + /// both sides are the deployed ones. + #[error("the policy and the window records it governs disagree: {0}")] + CombinedInvariant(String), // Both carry the driver's own account of what went wrong. Neither the // pool nor the driver repeats the connection string in its message, so // naming the cause costs no credential. @@ -234,6 +258,25 @@ pub struct ClaimRow { pub closed_at: Option>, } +impl ClaimRow { + /// Whether this claim holds its offering's terms still. This is the Rust + /// twin of the sentence `apply_policy` selects on: publication refuses to + /// change the service, location, mode or supply of an offering under a + /// claim that answers true here, so for such a claim the current policy + /// and the revision the claim names give the same terms. Under every + /// other claim, a closed one or a hold already past its expiry, the + /// offering is free to move and only the revision the claim names still + /// describes what was committed. + #[must_use] + pub fn pins_its_offering(&self, now: DateTime) -> bool { + self.state == ClaimState::Active + && match self.kind { + LedgerKind::Booking => true, + LedgerKind::Hold => self.hold_expires_at.is_some_and(|expiry| expiry > now), + } + } +} + #[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] #[serde(rename_all = "kebab-case")] pub enum ClaimState { @@ -642,6 +685,27 @@ impl PostgresStore { .await?; } transaction.commit().await?; + + let transaction = client.transaction().await?; + let applied: bool = transaction + .query_one( + "SELECT EXISTS(SELECT 1 FROM scheduling_schema_migrations WHERE version=$1)", + &[&DUPLICATE_LOOKUP_INDEX_MIGRATION_VERSION], + ) + .await? + .get(0); + if !applied { + transaction + .batch_execute(DUPLICATE_LOOKUP_INDEX_MIGRATION) + .await?; + transaction + .execute( + "INSERT INTO scheduling_schema_migrations(version,applied_at) VALUES($1,now()) ON CONFLICT(version) DO NOTHING", + &[&DUPLICATE_LOOKUP_INDEX_MIGRATION_VERSION], + ) + .await?; + } + transaction.commit().await?; Ok(()) } @@ -751,6 +815,11 @@ impl PostgresStore { if row.get::<_, String>(0) != scheduling_id { return Err(StoreError::Corrupt); } + // The candidate policy answers for the window records this + // deployment already holds. Every supply anchor and the meta row are + // locked here, and a records swap takes the same two in the same + // order, so the records read cannot move under this decision. + refuse_combined_conflicts(policy, &deployed_windows(&transaction).await?)?; let revision; let stored_revision = row.get::<_, i64>(1); let stored_digest = row.get::<_, String>(2); @@ -764,21 +833,16 @@ impl PostgresStore { ) .await? .and_then(|stored| stored.get(0)); - let Some(mut current_document) = current_document else { + let Some(current_document) = current_document else { return Err(StoreError::PolicyInUse( "the current policy document is unavailable; reapply the current policy before publishing a change" .to_owned(), )); }; - // Earlier branch builds retained mutable window records inside - // the policy document. Preserve that historical document and - // digest, but ignore its legacy member while comparing the - // lifecycle terms governed by the current policy shape. - if let Some(document) = current_document.as_object_mut() { - document.remove("windows"); - } - let current: SchedulingPolicy = - serde_json::from_value(current_document).map_err(|_| StoreError::Corrupt)?; + // The historical document and digest are preserved as they + // were published; only the lifecycle terms governed by the + // current policy shape are compared. + let current = retained_policy(current_document)?; let now = self.observed_now(); let active_offerings = transaction .query( @@ -847,13 +911,7 @@ impl PostgresStore { .await?; } for id in pool_ids { - transaction - .execute( - "INSERT INTO scheduling_supply(supply_id, kind) VALUES($1,$2) \ - ON CONFLICT(supply_id) DO NOTHING", - &[id, &"pool"], - ) - .await?; + anchor_supply(&transaction, id, "pool").await?; } transaction.commit().await?; Ok(revision) @@ -1026,6 +1084,32 @@ impl PostgresStore { row.map(map_claim_row).transpose() } + /// The offering as the policy revision `policy_revision` published it. + /// + /// Every claim names the revision it was committed under, and every + /// published revision retains its document. Publication may retire an + /// offering once no claim on it is live, so the close that retired it has + /// a receipt the current policy can no longer describe: the terms that + /// governed the claim are read from the revision that governed it. + pub async fn retained_offering( + &self, + policy_revision: i64, + offering: &str, + ) -> Result, StoreError> { + let client = self.client().await?; + let row = client + .query_opt( + "SELECT policy_document FROM scheduling_policy_revisions \ + WHERE policy_revision=$1", + &[&policy_revision], + ) + .await?; + let Some(document) = row.and_then(|row| row.get::<_, Option>(0)) else { + return Ok(None); + }; + Ok(retained_policy(document)?.offering(offering).cloned()) + } + /// The history of one claim, newest first, bounded by the listing limit. /// A page resumes strictly before the (occurred_at, event_id) pair its /// last row carries, so two events sharing an instant still page @@ -1176,6 +1260,7 @@ impl PostgresStore { &mut snapshot, offering, request.duplicate_key.as_deref(), + commitment.now, ) .await?; // The caller lock is taken after the supply lock, never before: every @@ -1288,6 +1373,7 @@ impl PostgresStore { &mut snapshot, offering, request.duplicate_key.as_deref(), + commitment.now, ) .await?; guard_revisions(&transaction, &commitment).await?; @@ -1619,6 +1705,7 @@ impl PostgresStore { &mut snapshot, offering, request.duplicate_key.as_deref(), + commitment.now, ) .await?; let appointment = transaction @@ -1887,7 +1974,8 @@ impl PostgresStore { let transaction = client.transaction().await?; let rows = transaction .query( - "UPDATE scheduling_claims SET state='expired', closed_at=now(), changed_at=now() \ + "UPDATE scheduling_claims SET state='expired', revision=revision + 1, \ + closed_at=now(), changed_at=now() \ WHERE claim_id IN (\ SELECT claim_id FROM scheduling_claims \ WHERE kind='hold' AND state='active' AND hold_expires_at <= $1 \ @@ -2393,13 +2481,22 @@ async fn lock_and_snapshot( /// Add the offering-wide duplicate fact that a time-bounded capacity snapshot /// may omit. The existing supply lock serializes this read with every writer -/// for the offering's frozen supply; the supporting partial index keeps the -/// lookup independent of the age or span of its openings. +/// for the offering's frozen supply; the supporting partial index carries the +/// offering and the booking's end beside the key, so the lookup stays +/// independent of the age or span of its openings and of the elapsed bookings +/// the party has accumulated under the same key. +/// +/// A booking counts while its own time has not passed. The guard is there so +/// one party holds one live booking per offering, and a booking that has +/// already happened is not one: there is no completion transition, and +/// cancellation closes at the cutoff, so an elapsed booking counted here would +/// refuse that party the offering permanently. async fn include_active_duplicate( transaction: &deadpool_postgres::Transaction<'_>, snapshot: &mut LedgerSnapshot, offering: ®istry_scheduling_core::OfferingPolicy, duplicate_key: Option<&str>, + now: DateTime, ) -> Result<(), StoreError> { if offering.duplicate_active_key.is_none() { return Ok(()); @@ -2412,8 +2509,9 @@ async fn include_active_duplicate( "SELECT claim_id, offering, supply_id, kind, channel, occupied_start, \ occupied_end, units, duplicate_key, hold_expires_at \ FROM scheduling_claims \ - WHERE state='active' AND kind='booking' AND offering=$1 AND duplicate_key=$2", - &[&offering.id, &duplicate_key], + WHERE state='active' AND kind='booking' AND offering=$1 AND duplicate_key=$2 \ + AND occupied_end > $3", + &[&offering.id, &duplicate_key, &now], ) .await?; for claim in snapshot_from_rows(rows).claims { @@ -2556,6 +2654,120 @@ async fn guard_revisions( Ok(()) } +/// Read a retained policy document. Earlier branch builds carried mutable +/// window records inside the document; that member left the policy shape, so +/// it is dropped rather than refused when an older document still carries it. +fn retained_policy(mut document: Value) -> Result { + if let Some(object) = document.as_object_mut() { + object.remove("windows"); + } + serde_json::from_value(document).map_err(|_| StoreError::Corrupt) +} + +/// The combined policy-by-records contradictions a database write refuses. +/// +/// `SchedulingPolicy::check_window_records` stays the single implementation +/// of the rules; this narrows its findings to the ones that are a genuine +/// disagreement between the two sides. Two of its reasons name an absence +/// instead: +/// +/// - `UnknownWindow` is a policy that references a window the deployment has +/// not published yet. That is the ordinary bootstrap order, and the runtime +/// already refuses each admission on such an offering loudly, so a write +/// must not be held hostage to it. +/// - `UnknownOffering` is a deployed window whose offering this publication +/// retires. Refusing it would make an arrival offering impossible to +/// withdraw, since the window records could only be removed afterwards. +/// +/// Everything that remains is a statement one side makes that the other +/// contradicts, and no ordering of the two writes makes it safe. In +/// particular `SharedSupplyUnpartitioned`, where a pool staffs a window and +/// also backs an exact-time offering: the two modes lock different anchors, +/// so their capacity transactions never serialize and the ledger cannot see +/// that it sold the same people twice. +fn combined_conflicts( + policy: &SchedulingPolicy, + windows: &[PublishedWindow], +) -> Vec { + policy + .check_window_records(windows) + .into_iter() + .filter(|finding| { + !matches!( + finding.reason, + PolicyCheckReason::UnknownWindow | PolicyCheckReason::UnknownOffering + ) + }) + .collect() +} + +/// Anchor one supply identifier under the kind that claims it. +/// +/// The pool anchors and the window anchors are written by two independent +/// publish paths into one table keyed by the identifier, and the two +/// namespaces are authored separately. Both paths write it through here, so +/// an identifier the other kind already holds is refused by name at the +/// second of them rather than aborting the transaction on the primary key or +/// silently leaving the other kind's row standing. A row of the same kind is +/// left exactly as it is: republishing a policy re-anchors its pools, and +/// that is a no-op, not a conflict. +async fn anchor_supply( + transaction: &deadpool_postgres::Transaction<'_>, + supply_id: &str, + kind: &str, +) -> Result<(), StoreError> { + let anchored: String = transaction + .query_one( + "INSERT INTO scheduling_supply(supply_id, kind) VALUES($1,$2) \ + ON CONFLICT(supply_id) DO UPDATE SET kind=scheduling_supply.kind \ + RETURNING kind", + &[&supply_id, &kind], + ) + .await? + .get(0); + if anchored != kind { + return Err(StoreError::SupplyIdentifierCollision(format!( + "{supply_id} already anchors {anchored} supply and cannot also anchor a {kind}" + ))); + } + Ok(()) +} + +/// The window records the deployment currently holds, in identifier order. +async fn deployed_windows( + transaction: &deadpool_postgres::Transaction<'_>, +) -> Result, StoreError> { + transaction + .query( + "SELECT window_record FROM scheduling_windows ORDER BY window_id", + &[], + ) + .await? + .iter() + .map(|row| { + serde_json::from_value::(row.get(0)).map_err(|_| StoreError::Corrupt) + }) + .collect() +} + +/// The refusal carrying every contradiction, or `Ok(())` when there is none. +fn refuse_combined_conflicts( + policy: &SchedulingPolicy, + windows: &[PublishedWindow], +) -> Result<(), StoreError> { + let conflicts = combined_conflicts(policy, windows); + if conflicts.is_empty() { + return Ok(()); + } + Err(StoreError::CombinedInvariant( + conflicts + .iter() + .map(ToString::to_string) + .collect::>() + .join(", "), + )) +} + /// Replay a stored attempt: the same key with a different payload is /// refused as reused, an erased receipt as expired, and a retained one is /// answered exactly as it was. @@ -2693,32 +2905,44 @@ pub(crate) async fn replace_facts_in_transaction( &[], ) .await?; - let stored_id: String = transaction + let meta = transaction .query_one( - "SELECT scheduling_id FROM scheduling_meta WHERE singleton FOR UPDATE", + "SELECT scheduling_id, policy_revision, policy_digest FROM scheduling_meta \ + WHERE singleton FOR UPDATE", &[], ) - .await? - .get(0); - if stored_id != scheduling_id { + .await?; + if meta.get::<_, String>(0) != scheduling_id { return Err(StoreError::DeploymentIdentity); } + // The incoming records answer to the policy deployed now, the same pair + // the authoring tool checks together offline. An empty digest is a + // deployment whose records are being seeded before any policy exists, + // and there is nothing yet for them to contradict. + let policy_digest: String = meta.get(2); + if !policy_digest.is_empty() { + let policy_revision: i64 = meta.get(1); + let document: Option = transaction + .query_opt( + "SELECT policy_document FROM scheduling_policy_revisions \ + WHERE policy_revision=$1 AND policy_digest=$2", + &[&policy_revision, &policy_digest], + ) + .await? + .and_then(|stored| stored.get(0)); + let Some(document) = document else { + return Err(StoreError::CombinedInvariant( + "the deployed policy document is unavailable; reapply the current policy before replacing the records" + .to_owned(), + )); + }; + refuse_combined_conflicts(&retained_policy(document)?, &facts.windows)?; + } let occupied = occupied_resources_changed_by(transaction, facts).await?; if !occupied.is_empty() { return Err(StoreError::FactsInUse(occupied.join(", "))); } - let current_windows = transaction - .query( - "SELECT window_record FROM scheduling_windows ORDER BY window_id", - &[], - ) - .await? - .iter() - .map(|row| { - serde_json::from_value::(row.get(0)) - .map_err(|_| StoreError::Corrupt) - }) - .collect::, _>>()?; + let current_windows = deployed_windows(transaction).await?; let revision_heads = transaction .query( "SELECT window_id, window_record FROM scheduling_window_revision_heads ORDER BY window_id", @@ -2846,12 +3070,7 @@ pub(crate) async fn replace_facts_in_transaction( &[&window.id, &record], ) .await?; - transaction - .execute( - "INSERT INTO scheduling_supply(supply_id, kind) VALUES($1,'window')", - &[&window.id], - ) - .await?; + anchor_supply(transaction, &window.id, "window").await?; transaction .execute( "INSERT INTO scheduling_window_revision_heads(window_id, window_record, updated_at) \ @@ -3067,7 +3286,13 @@ impl CapacityStatements for deadpool_postgres::Transaction<'_> { ) .await?; if rows.len() != supply_ids.len() { - return Err(StoreError::Corrupt); + let anchored: Vec = rows.iter().map(|row| row.get(0)).collect(); + let missing: Vec<&str> = supply_ids + .iter() + .map(String::as_str) + .filter(|wanted| !anchored.iter().any(|found| found == wanted)) + .collect(); + return Err(StoreError::UnanchoredSupply(missing.join(", "))); } Ok(()) } diff --git a/crates/registry-scheduling/tests/postgres_commitments.rs b/crates/registry-scheduling/tests/postgres_commitments.rs index 66a973d9f..23ca1ffef 100644 --- a/crates/registry-scheduling/tests/postgres_commitments.rs +++ b/crates/registry-scheduling/tests/postgres_commitments.rs @@ -42,7 +42,7 @@ use registry_scheduling_core::{ location_closure_intervals, location_open_intervals, parse_policy_yaml, AdmissionRefusal, AdmissionRequest, CalendarExceptionRecord, Channel, ExceptionRecordKind, LocationRecord, OfferingPolicy, PartyCounts, PoolMember, PublishedWindow, RequiredUnitsPolicy, ResourcePool, - SchedulingFacts, SchedulingPolicy, WindowSubquota, + SchedulingFacts, SchedulingPolicy, WindowStaffing, WindowSubquota, }; use serde_json::{json, Value}; use std::collections::BTreeMap; @@ -1416,6 +1416,123 @@ async fn duplicate_active_keys_are_scoped_to_the_offering() { assert_eq!(status, StatusCode::CREATED, "{appointment}"); } +/// The duplicate guard exists so one party holds one live booking per offering +/// at a time. A booking whose time has passed is not live, and counting it +/// would refuse that party forever: the product publishes no completion +/// transition, and cancellation closes at the cutoff before the appointment +/// even starts, so nothing the party can do would ever release the key. +#[tokio::test] +async fn an_elapsed_booking_no_longer_holds_the_partys_duplicate_key() { + let keyed = POLICY.replacen( + " requiresCapabilities: []", + " duplicateActiveKey: subject\n requiresCapabilities: []", + 2, + ); + let fx = fixture_publishing( + &keyed, + &["north-counter".to_owned(), "two-counter".to_owned()], + ) + .await; + + let first = first_slot(&fx, OFFERING, 300, 440).await; + let mut first_admission = admission(&fx, OFFERING, first); + first_admission["duplicateKey"] = json!("subject:elapsed"); + let (status, appointment) = fx + .post( + "/v1/appointments", + &fx.agent, + "duplicate-elapsed-first", + json!({"hold": null, "admission": first_admission}), + ) + .await; + assert_eq!(status, StatusCode::CREATED, "{appointment}"); + + let second = first_slot(&fx, OFFERING, 600, 740).await; + let mut second_admission = admission(&fx, OFFERING, second); + second_admission["duplicateKey"] = json!("subject:elapsed"); + let (status, problem) = fx + .post( + "/v1/appointments", + &fx.agent, + "duplicate-elapsed-live", + json!({"hold": null, "admission": second_admission.clone()}), + ) + .await; + assert_eq!(status, StatusCode::CONFLICT, "{problem}"); + assert_eq!( + problem["code"], "booking.duplicate-active", + "a live booking holds the key" + ); + + // What the passage of time does to that booking, with no sweeper and no + // transition the party could have reached. + fx.admin + .execute( + "UPDATE scheduling_claims SET \ + displayed_start = now() - interval '2 hours', \ + displayed_end = now() - interval '1 hour', \ + occupied_start = now() - interval '2 hours', \ + occupied_end = now() - interval '1 hour' \ + WHERE duplicate_key = $1", + &[&"subject:elapsed"], + ) + .await + .expect("the standing booking elapses"); + + let (status, appointment) = fx + .post( + "/v1/appointments", + &fx.agent, + "duplicate-elapsed-after", + json!({"hold": null, "admission": second_admission}), + ) + .await; + assert_eq!( + status, + StatusCode::CREATED, + "an elapsed booking no longer holds the key: {appointment}" + ); +} + +/// The guard above reads every claim that carries one party's key, and a +/// booking stays `active` once its time has passed, so the set it filters +/// accumulates for the life of the deployment. Hold the schema to an index +/// that answers the whole predicate: on the key alone the offering and the +/// elapsed bookings fall to a recheck over that accumulating history. +#[tokio::test] +async fn the_duplicate_guard_is_indexed_on_the_whole_predicate_it_filters() { + let fx = fixture().await; + + let definitions: Vec = fx + .admin + .query( + "SELECT indexdef FROM pg_indexes \ + WHERE schemaname=current_schema() AND tablename='scheduling_claims' \ + AND indexdef LIKE '%duplicate_key%'", + &[], + ) + .await + .expect("read the claim ledger indexes") + .iter() + .map(|row| row.get(0)) + .collect(); + + assert_eq!( + definitions.len(), + 1, + "one index answers the duplicate guard: {definitions:?}" + ); + let definition = &definitions[0]; + assert!( + definition.contains("(duplicate_key, offering, occupied_end)"), + "the index carries every column the guard filters on: {definition}" + ); + assert!( + definition.contains("state = 'active'") && definition.contains("duplicate_key IS NOT NULL"), + "the index stays partial to the live keyed claims: {definition}" + ); +} + /// A policy may move an exact-time offering's opening dates while retaining /// its offering and pool. The standing appointment then falls outside the new /// opening span, but its offering-scoped duplicate key remains active and must @@ -1922,6 +2039,59 @@ async fn an_expired_hold_returns_capacity_and_refuses_confirmation() { assert_eq!(problem["code"], "hold.released"); } +/// Every closing writer advances the claim's revision, so the history event +/// that closes a claim is distinguishable from the one that opened it. The +/// expiry sweeper closes a hold exactly as release and cancellation do, and +/// must leave the same trace behind it. +#[tokio::test] +async fn hold_expiry_advances_the_claim_revision_like_every_other_close() { + let fx = fixture().await; + let slot = first_slot(&fx, OFFERING, 90, 200).await; + let (status, hold) = fx + .post( + "/v1/holds", + &fx.agent, + "hold-revision", + admission(&fx, OFFERING, slot), + ) + .await; + assert_eq!(status, StatusCode::CREATED); + let hold_id = Uuid::parse_str(hold["holdId"].as_str().unwrap()).expect("a hold UUID"); + + let expired = fx + .store + .expire_due_holds(Utc::now() + TimeDelta::minutes(11), 100) + .await + .expect("the hold expiry pass"); + assert_eq!(expired, 1); + + let history = fx + .store + .claim_history(hold_id, None, 10) + .await + .expect("the hold history"); + let revision_of = |kind: &str| { + history + .iter() + .find(|event| event["kind"] == kind) + .and_then(|event| event["revision"].as_i64()) + }; + assert_eq!(revision_of("held"), Some(1)); + assert_eq!( + revision_of("expired"), + Some(2), + "the expiry event carries the revision its own close minted" + ); + + let claim = fx + .store + .claim(hold_id) + .await + .expect("the claim read") + .expect("the expired hold row"); + assert_eq!(claim.revision, 2); +} + /// SEC-01 and SEC-11 at their shared transaction boundary. A confirmation may /// enter while its hold is live and pause before it reaches the supply anchor. /// Once the hold expires, a later transaction may reclaim and book that @@ -4325,6 +4495,221 @@ async fn records_replacement_refuses_to_move_a_window_with_standing_commitments( assert!(refusal.to_string().contains(WINDOW_ID), "{refusal}"); } +/// A pool that staffs a window may not also back an exact-time offering: the +/// two modes lock different anchors, so their capacity transactions never +/// serialize against each other and the ledger cannot see that it sold the +/// same people twice. The authoring check refuses that combination against a +/// policy file, offline. The database write is the other place it has to +/// hold, because an operator reaching `replace_facts` never passed the file +/// through that tool. +#[tokio::test] +async fn records_replacement_refuses_a_window_staffed_by_an_exact_time_pool() { + let start = (Utc::now() + TimeDelta::days(2)) + .with_nanosecond(0) + .expect("second precision"); + let fx = fixture_publishing( + &policy_with_window(), + &["north-counter".to_owned(), "two-counter".to_owned()], + ) + .await; + + let mut shared = records_with_window(start); + shared.windows[0].staffing = Some(WindowStaffing { + pool: "north-counter".to_owned(), + reserved_members: None, + because: "test".to_owned(), + }); + let refusal = fx + .store + .replace_facts(SCHEDULING_ID, &shared, Uuid::new_v4(), operator_audit()) + .await + .expect_err("a window may not be staffed by a pool an exact-time offering sells"); + assert!( + refusal + .to_string() + .contains("windows[0].staffing.pool: shared-supply-unpartitioned"), + "{refusal}" + ); + + let (records, _) = fx + .store + .facts() + .await + .expect("the records after the refusal"); + assert!( + records.windows.is_empty(), + "a refused swap writes no window record" + ); +} + +/// A resource pool and a published window anchor their capacity +/// transactions on one row keyed by the identifier, and the two namespaces +/// are authored separately. Sharing an identifier makes the two publish +/// paths write the same row: the records write would abort on the primary +/// key, and the policy write used to skip the anchor silently, leaving the +/// pool serializing against a window row that the next records swap deletes. +/// Both directions are refused by name instead. +#[tokio::test] +async fn a_supply_identifier_may_not_anchor_both_a_pool_and_a_window() { + let start = (Utc::now() + TimeDelta::days(2)) + .with_nanosecond(0) + .expect("second precision"); + let fx = fixture_publishing( + &policy_with_window().replace(WINDOW_ID, "north-counter"), + &["north-counter".to_owned(), "two-counter".to_owned()], + ) + .await; + + // A window record taking the identifier a live pool already anchors. + let mut colliding = records_with_window(start); + colliding.windows[0].id = "north-counter".to_owned(); + let refusal = fx + .store + .replace_facts(SCHEDULING_ID, &colliding, Uuid::new_v4(), operator_audit()) + .await + .expect_err("a window may not take a pool's supply identifier"); + assert!( + refusal + .to_string() + .contains("windows[0].id: supply-identifier-collision"), + "{refusal}" + ); + + // The reverse, taken at the anchor row itself: the row is planted here + // because both writes now refuse to produce one, and the publication + // that would claim it as a pool is refused rather than skipping the + // anchor and stranding the offering that needs it. + fx.admin + .execute( + "INSERT INTO scheduling_supply(supply_id, kind) VALUES('south-counter','window')", + &[], + ) + .await + .expect("plant a window anchor the policy will collide with"); + let deployed = fx + .store + .scheduling_meta() + .await + .expect("the deployment metadata before the refusal"); + let mut replacement = + parse_policy_yaml(&policy_with_window().replace(WINDOW_ID, "north-counter")) + .expect("the deployed policy"); + replacement.scheduling.version += 1; + replacement + .offerings + .iter_mut() + .find(|offering| offering.id == "registry-review-45") + .and_then(|offering| offering.exact_time.as_mut()) + .expect("the idle exact-time offering") + .pool = "south-counter".to_owned(); + let digest = replacement.policy_digest(); + let refusal = fx + .store + .apply_policy( + SCHEDULING_ID, + &digest, + &[ + "north-counter".to_owned(), + "two-counter".to_owned(), + "south-counter".to_owned(), + ], + &replacement, + ) + .await + .expect_err("a pool may not take a window's supply identifier"); + assert!(refusal.to_string().contains("south-counter"), "{refusal}"); + + let after = fx + .store + .scheduling_meta() + .await + .expect("the deployment metadata after the refusal"); + assert_eq!(after, deployed, "a refused publication deploys nothing"); + let anchored: String = fx + .admin + .query_one( + "SELECT kind FROM scheduling_supply WHERE supply_id='south-counter'", + &[], + ) + .await + .expect("read the planted anchor") + .get(0); + assert_eq!(anchored, "window", "the standing anchor keeps its kind"); +} + +/// The same combined invariant, reached from the other side. The deployed +/// records are legal under the deployed policy, and it is the next policy +/// publication that puts an exact-time offering on the pool staffing a +/// standing window. Fixing only one of the two writes leaves this direction +/// open. +#[tokio::test] +async fn policy_publication_refuses_an_exact_time_offering_on_a_deployed_windows_staffing() { + let start = (Utc::now() + TimeDelta::days(2)) + .with_nanosecond(0) + .expect("second precision"); + let pools = [ + "north-counter".to_owned(), + "two-counter".to_owned(), + "south-counter".to_owned(), + ]; + let fx = fixture_publishing(&policy_with_window(), &pools).await; + + // No exact-time offering sells `south-counter` under the deployed + // policy, so staffing the window from it is admissible today. + let mut staffed = records_with_window(start); + staffed.pools.push(ResourcePool { + id: "south-counter".to_owned(), + members: vec![PoolMember { + resource_id: "station-3".to_owned(), + capabilities: Vec::new(), + available: true, + }], + }); + staffed.windows[0].staffing = Some(WindowStaffing { + pool: "south-counter".to_owned(), + reserved_members: None, + because: "test".to_owned(), + }); + fx.store + .replace_facts(SCHEDULING_ID, &staffed, Uuid::new_v4(), operator_audit()) + .await + .expect("a window staffed by a pool no exact-time offering sells"); + + let deployed = fx + .store + .scheduling_meta() + .await + .expect("the deployment metadata before the refusal"); + + let conflicting = parse_policy_yaml( + &policy_with_window().replace("pool: two-counter", "pool: south-counter"), + ) + .expect("a policy moving an idle offering onto the window's staffing"); + let digest = conflicting.policy_digest(); + let refusal = fx + .store + .apply_policy(SCHEDULING_ID, &digest, &pools, &conflicting) + .await + .expect_err("an exact-time offering may not take a standing window's staffing"); + assert!( + refusal + .to_string() + .contains("windows[0].staffing.pool: shared-supply-unpartitioned"), + "{refusal}" + ); + + let after = fx + .store + .scheduling_meta() + .await + .expect("the deployment metadata after the refusal"); + assert_eq!(after, deployed, "a refused publication deploys nothing"); + assert_ne!( + after.2, digest, + "the conflicting policy is not the deployed one" + ); +} + #[tokio::test] async fn changed_window_records_must_advance_the_revision_even_after_removal() { let start = (Utc::now() + TimeDelta::days(2)) @@ -4456,6 +4841,37 @@ async fn changed_window_records_must_advance_the_revision_even_after_removal() { .expect("an exact unchanged reapply remains accepted"); } +/// Write one window record and its supply anchor straight into the tables, +/// underneath `replace_facts` and the invariants it holds. A record that +/// contradicts the deployed policy cannot be written through the store any +/// more, so this is how a test still reaches the shape an older runtime, or +/// a hand-edited table, can leave behind. +async fn plant_window(fx: &Fixture, window: &PublishedWindow) { + let record = serde_json::to_value(window).expect("a serializable window record"); + fx.admin + .execute( + "INSERT INTO scheduling_windows(window_id, window_record) VALUES($1,$2) \ + ON CONFLICT(window_id) DO UPDATE SET window_record=EXCLUDED.window_record", + &[&window.id, &record], + ) + .await + .expect("plant the window record"); + fx.admin + .execute( + "INSERT INTO scheduling_supply(supply_id, kind) VALUES($1,'window') \ + ON CONFLICT(supply_id) DO NOTHING", + &[&window.id], + ) + .await + .expect("plant the window's supply anchor"); +} + +/// The per-admission check is defense in depth, not the only defense: both +/// database writes refuse to produce a window record that contradicts the +/// policy governing it. The records here are planted underneath those +/// writes, so a deployment that carries such a pair from anywhere else still +/// refuses every admission on it rather than committing capacity it cannot +/// account for. #[tokio::test] async fn a_window_record_must_belong_to_the_authorized_offering_and_location() { let start = (Utc::now() + TimeDelta::hours(3)) @@ -4466,15 +4882,7 @@ async fn a_window_record_must_belong_to_the_authorized_offering_and_location() { &["north-counter".to_owned(), "two-counter".to_owned()], ) .await; - fx.store - .replace_facts( - SCHEDULING_ID, - &records_with_window(start), - Uuid::new_v4(), - operator_audit(), - ) - .await - .expect("seed the intentionally misbound operator record"); + plant_window(&fx, &records_with_window(start).windows[0]).await; let caller = agent_token_for_service("registry-review"); let mut request = arrival(&fx, start, None); request["admission"]["offering"] = json!(FOREIGN_WINDOW_OFFERING); @@ -4494,15 +4902,7 @@ async fn a_window_record_must_belong_to_the_authorized_offering_and_location() { wrong_location.windows[0].revision = WINDOW_REVISION + 1; wrong_location.windows[0].offering = FOREIGN_WINDOW_OFFERING.to_owned(); wrong_location.windows[0].location = "two-counter".to_owned(); - fx.store - .replace_facts( - SCHEDULING_ID, - &wrong_location, - Uuid::new_v4(), - operator_audit(), - ) - .await - .expect("seed the intentionally wrong-location operator record"); + plant_window(&fx, &wrong_location.windows[0]).await; request["admission"]["windowRevision"] = json!(WINDOW_REVISION + 1); let (status, problem) = fx .post( @@ -4602,6 +5002,205 @@ async fn policy_publication_refuses_to_remove_an_offering_with_a_live_appointmen assert_eq!(cancelled["state"], "cancelled"); } +/// A second edge over the same deployment, serving `policy` as the current +/// one. Everything else stays the fixture's: the same store, schema, hooks, +/// and audit keying, so a request through it meets the claims the first edge +/// committed. +async fn republished(fx: &Fixture, policy: SchedulingPolicy, pool_ids: &[String]) -> Router { + let digest = policy.policy_digest(); + let revision = fx + .store + .apply_policy(SCHEDULING_ID, &digest, pool_ids, &policy) + .await + .expect("republish the scheduling policy"); + let keying = AuditHashSecret::new(vec![0x42; 32]).expect("the test audit hash secret"); + let service = Arc::new( + SchedulingService::new( + fx.store.clone(), + policy, + SCHEDULING_ID.to_owned(), + revision, + digest, + AuditKeyHasher::Keyed(keying), + 7, + ) + .with_hooks(fx.hooks.clone()), + ); + router(HttpState { + service, + authenticator: Arc::new(authenticator()), + store: fx.store.clone(), + }) +} + +/// An offering stays in the policy while any claim on it is live, and may be +/// retired once every one of them is closed. The receipts those closures wrote +/// outlive the offering, and the retry each receipt exists to serve must still +/// replay its recorded answer rather than read as an outage. +#[tokio::test] +async fn a_retry_replays_its_receipt_after_its_offering_leaves_the_policy() { + let fx = fixture().await; + let (appointment_id, revision) = booked(&fx, 300, 440, "retired-offering-create").await; + let slot = first_slot(&fx, OFFERING, 600, 740).await; + let (status, hold) = fx + .post( + "/v1/holds", + &fx.agent, + "retired-offering-hold", + admission(&fx, OFFERING, slot), + ) + .await; + assert_eq!(status, StatusCode::CREATED, "{hold}"); + let hold_id = hold["holdId"].as_str().expect("a hold id").to_owned(); + + let (status, _) = fx.delete(&format!("/v1/holds/{hold_id}"), &fx.agent).await; + assert_eq!(status, StatusCode::NO_CONTENT); + let cancellation = json!({"observedRevision": revision, "reason": null}); + let (status, cancelled) = fx + .post( + &format!("/v1/appointments/{appointment_id}/cancel"), + &fx.agent, + "retired-offering-cancel", + cancellation.clone(), + ) + .await; + assert_eq!(status, StatusCode::OK, "{cancelled}"); + + let mut replacement = parse_policy_yaml(POLICY).expect("the current policy"); + replacement.scheduling.version += 1; + replacement + .offerings + .retain(|offering| offering.id != OFFERING); + let retired = republished( + &fx, + replacement, + &["north-counter".to_owned(), "two-counter".to_owned()], + ) + .await; + + let (status, replayed) = send( + retired.clone(), + "DELETE".to_owned(), + format!("/v1/holds/{hold_id}"), + fx.agent.clone(), + None, + None, + ) + .await; + assert_eq!( + status, + StatusCode::NO_CONTENT, + "the release replays its receipt: {replayed}" + ); + + let (status, replayed) = send( + retired, + "POST".to_owned(), + format!("/v1/appointments/{appointment_id}/cancel"), + fx.agent.clone(), + Some("retired-offering-cancel".to_owned()), + Some(cancellation), + ) + .await; + assert_eq!( + status, + StatusCode::OK, + "the cancellation replays its receipt: {replayed}" + ); + assert_eq!(replayed["state"], "cancelled"); + assert_eq!(replayed["appointmentId"], appointment_id); +} + +/// The same retry, from the other side of publication. Retirement is not the +/// only change a closed claim outlives: once nothing on an offering is live, +/// publication may also keep its identifier and change the service or the +/// location it is sold under, because only a live claim pins those. The +/// receipt a closed claim carries was authorized against the offering as it +/// stood when the claim was committed, so its retry has to be judged the same +/// way. Reading the live policy instead would refuse a caller holding the very +/// grant that committed the claim, and would let a grant over the offering's +/// new service reach a receipt written under the old one. +#[tokio::test] +async fn a_retry_replays_its_receipt_after_its_offering_changes_service() { + let fx = fixture().await; + let (appointment_id, revision) = booked(&fx, 300, 440, "moved-offering-create").await; + let slot = first_slot(&fx, OFFERING, 600, 740).await; + let (status, hold) = fx + .post( + "/v1/holds", + &fx.agent, + "moved-offering-hold", + admission(&fx, OFFERING, slot), + ) + .await; + assert_eq!(status, StatusCode::CREATED, "{hold}"); + let hold_id = hold["holdId"].as_str().expect("a hold id").to_owned(); + + let (status, _) = fx.delete(&format!("/v1/holds/{hold_id}"), &fx.agent).await; + assert_eq!(status, StatusCode::NO_CONTENT); + let cancellation = json!({"observedRevision": revision, "reason": null}); + let (status, cancelled) = fx + .post( + &format!("/v1/appointments/{appointment_id}/cancel"), + &fx.agent, + "moved-offering-cancel", + cancellation.clone(), + ) + .await; + assert_eq!(status, StatusCode::OK, "{cancelled}"); + + // The booking agent holds every action over registry-update and + // appointment.create alone over registry-review, so an offering that + // changes service leaves the grant which committed both of these claims + // unable to close either one again. + let mut replacement = parse_policy_yaml(POLICY).expect("the current policy"); + replacement.scheduling.version += 1; + replacement + .offerings + .iter_mut() + .find(|offering| offering.id == OFFERING) + .expect("the exact-time offering") + .service = "registry-review".to_owned(); + let moved = republished( + &fx, + replacement, + &["north-counter".to_owned(), "two-counter".to_owned()], + ) + .await; + + let (status, replayed) = send( + moved.clone(), + "DELETE".to_owned(), + format!("/v1/holds/{hold_id}"), + fx.agent.clone(), + None, + None, + ) + .await; + assert_eq!( + status, + StatusCode::NO_CONTENT, + "the release replays its receipt: {replayed}" + ); + + let (status, replayed) = send( + moved, + "POST".to_owned(), + format!("/v1/appointments/{appointment_id}/cancel"), + fx.agent.clone(), + Some("moved-offering-cancel".to_owned()), + Some(cancellation), + ) + .await; + assert_eq!( + status, + StatusCode::OK, + "the cancellation replays its receipt: {replayed}" + ); + assert_eq!(replayed["state"], "cancelled"); + assert_eq!(replayed["appointmentId"], appointment_id); +} + #[tokio::test] async fn policy_publication_refuses_to_move_an_active_offering_between_pools() { let fx = fixture().await; diff --git a/docs/site/src/data/scheduling-api.yaml b/docs/site/src/data/scheduling-api.yaml index 74740f4da..ab3766914 100644 --- a/docs/site/src/data/scheduling-api.yaml +++ b/docs/site/src/data/scheduling-api.yaml @@ -3,7 +3,7 @@ columns: [Code, Status, When] rows: - ['`authentication.refused`', '401', 'The bearer credential is missing, invalid, or expired. Sign in again.'] - - ['`booking.duplicate-active`', '409', 'An active booking already holds this party''s duplicate key. Cancel or complete it before booking again.'] + - ['`booking.duplicate-active`', '409', 'An active booking already holds this party''s duplicate key. Cancel it, or wait until it has passed, before booking again.'] - ['`cancellation.cutoff-passed`', '409', 'The cancellation cutoff for this appointment has passed.'] - ['`capability.unmatched`', '422', 'No backing member carries every capability the offering requires.'] - ['`capacity.exhausted`', '409', 'The supply is fully committed for the requested interval. Choose another time.'] diff --git a/products/identifiers/generated/catalog.v1.json b/products/identifiers/generated/catalog.v1.json index a0499fd14..409589f22 100644 --- a/products/identifiers/generated/catalog.v1.json +++ b/products/identifiers/generated/catalog.v1.json @@ -2095,7 +2095,7 @@ "description": "The bearer credential is missing, invalid, or expired. Sign in again.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "authentication.refused", @@ -2111,10 +2111,10 @@ "compatibilityLine": "v1alpha1", "owner": "scheduling", "title": "Duplicate active booking", - "description": "An active booking already holds this party's duplicate key. Cancel or complete it before booking again.", + "description": "An active booking already holds this party's duplicate key. Cancel it, or wait until it has passed, before booking again.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "booking.duplicate-active", @@ -2133,7 +2133,7 @@ "description": "The cancellation cutoff for this appointment has passed, so it can no longer be cancelled.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "cancellation.cutoff-passed", @@ -2152,7 +2152,7 @@ "description": "The party or backing supply does not carry every capability this offering requires.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "capability.unmatched", @@ -2171,7 +2171,7 @@ "description": "The supply is fully committed for the requested interval. Choose another time.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "capacity.exhausted", @@ -2190,7 +2190,7 @@ "description": "This cursor has expired. Start again without a cursor and deduplicate entries by id.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "cursor.expired", @@ -2209,7 +2209,7 @@ "description": "The cursor is invalid for this request.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "cursor.invalid", @@ -2228,7 +2228,7 @@ "description": "A required eligibility check could not run. Retry; do not treat this as permission.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "eligibility.unavailable", @@ -2247,7 +2247,7 @@ "description": "The hold expired before confirmation. Its capacity is bookable again; start a new request.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "hold.expired", @@ -2266,7 +2266,7 @@ "description": "The claim is not an active hold. It may already be confirmed or released.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "hold.released", @@ -2285,7 +2285,7 @@ "description": "A required lifecycle hook could not run, so the request was refused rather than half-applied.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "hook.unavailable", @@ -2304,7 +2304,7 @@ "description": "The requested start is earlier than the lead time allows or further ahead than the horizon allows.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "horizon.outside", @@ -2323,7 +2323,7 @@ "description": "The stored response for this idempotency key has expired. Reconcile the original operation before choosing a new key.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "idempotency.expired", @@ -2342,7 +2342,7 @@ "description": "This idempotency key was used for a different request.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "idempotency.key-reused", @@ -2361,7 +2361,7 @@ "description": "A closure covers the requested start. Choose a start outside the closure.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "location.closed", @@ -2380,7 +2380,7 @@ "description": "Your current Scheduling authority does not allow this operation.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "operation.not-authorized", @@ -2399,7 +2399,7 @@ "description": "The party is larger than this offering can ever serve, or its size falls outside the published units.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "party.capacity-inadequate", @@ -2418,7 +2418,7 @@ "description": "The policy revision changed before the request committed. Reload the catalogue and try again with the current revision.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "policy.changed", @@ -2437,7 +2437,7 @@ "description": "The appointment changed since you loaded it. Reload and try again.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "precondition.failed", @@ -2456,7 +2456,7 @@ "description": "This mutation requires the revision you loaded, or the duplicate key this offering keys on.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "precondition.required", @@ -2475,7 +2475,7 @@ "description": "The party is missing a prerequisite this offering requires.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "prerequisite.missing", @@ -2494,7 +2494,7 @@ "description": "The selected Scheduling profile does not authorize this request.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "profile.not-authorized", @@ -2513,7 +2513,7 @@ "description": "The request body exceeds the accepted size.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "request.body-too-large", @@ -2532,7 +2532,7 @@ "description": "The request could not be read as a Scheduling request.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "request.invalid", @@ -2551,7 +2551,7 @@ "description": "The route exists but not for this method.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "request.method-not-allowed", @@ -2570,7 +2570,7 @@ "description": "The requested route does not exist.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "request.not-found", @@ -2589,7 +2589,7 @@ "description": "The request body could not be processed.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "request.unprocessable", @@ -2608,7 +2608,7 @@ "description": "The request body is not JSON.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "request.unsupported-media-type", @@ -2627,7 +2627,7 @@ "description": "Every capable member is unavailable for this interval.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "resource.unavailable", @@ -2646,7 +2646,7 @@ "description": "The window revision changed before the request committed. Reload the catalogue and try again.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "revision.mismatch", @@ -2665,7 +2665,7 @@ "description": "No published schedule serves that start. Choose a start on the published grid.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "schedule.unpublished", @@ -2684,7 +2684,7 @@ "description": "Scheduling storage is unavailable. Try again after the service recovers.", "source": { "path": "crates/registry-scheduling-core/src/problem.rs", - "sha256": "8e72874fb179d87c3d8f158faff5c7e78f48b72c9a59adf3f7697f70606c58fc" + "sha256": "58f980f92c75d7bcc6d0fc41bb9945cb69609fe4af9efc570f936240bdb8244b" }, "problem": { "code": "service.unavailable", diff --git a/products/scheduling/CHANGELOG.md b/products/scheduling/CHANGELOG.md index b25bd9cda..79f519af8 100644 --- a/products/scheduling/CHANGELOG.md +++ b/products/scheduling/CHANGELOG.md @@ -10,11 +10,17 @@ appointment is created, moved, or released only inside that transaction, and no other product may write the ledger; eligibility stays with the source system. -- Validate a policy against the published window records it governs in the - authoring tooling, which refuses a window whose staffing pool also backs an - exact-time offering. Policy publication and records replacement do not - repeat that combined check at the database, so a deployment that bypasses - the authoring tooling can still publish the combination it refuses. +- Validate a policy against the published window records it governs, which + refuses a window whose staffing pool also backs an exact-time offering. + The authoring tooling reports it against the files on disk, and policy + publication and records replacement re-run the same check at the database + under the locks they already hold, so a deployment that bypasses the + authoring tooling cannot publish the combination it refuses. +- Keep one supply identifier to one kind of supply. A resource pool and a + published window anchor their capacity transactions on the same row keyed by + that identifier, so authoring refuses the collision from either side and both + anchor writes name the standing supply rather than aborting one publish path + and silently skipping the anchor on the other. - Authorize every commitment with a task grant whose scheduling bounds name the offering's service, its location, and the action, bounded to 64 permissions of 32 actions with no wildcard. Only the grant's expiry is @@ -35,6 +41,16 @@ - Retain idempotency attempt receipts for the configured period and forget listing cursors after fifteen minutes. Appointment, history, outbox, and audit retention are deferred. +- Answer a release or cancellation retry from the policy revision that + governed the claim. Publication may retire an offering, or keep its + identifier and sell it under a different service or location, once no claim + on it is live, and the retry each receipt exists to serve is still + authorized against the offering as it stood when the claim was committed. +- Bound the offering-wide duplicate guard to the bookings it decides on. The + ledger closes no booking when its time passes, so the active claims carrying + one party's key accumulate for the life of the deployment; the guard now + reads an index over the key, the offering, and the booking's end rather than + over the key alone. - Provide `schedulingctl init`, `check`, `test`, `explain`, and `package`, and write complete `runtime.example.yaml` and `records.yaml` documents beside every initialized project. diff --git a/products/scheduling/SECURITY-REVIEW-NOTES.md b/products/scheduling/SECURITY-REVIEW-NOTES.md index 5e549bf1f..a405fecd4 100644 --- a/products/scheduling/SECURITY-REVIEW-NOTES.md +++ b/products/scheduling/SECURITY-REVIEW-NOTES.md @@ -259,10 +259,91 @@ request shapes stay strict. *Tests:* `a_reworded_title_or_detail_is_still_the_problem_the_code_names` (`crates/registry-scheduling-client/src/error.rs`). +## The combined policy and records invariants at the database + +**Threat:** a deployment can come to hold a policy and window records +that contradict each other. The consequential case is a pool that staffs +a published window and also backs an exact-time offering: an exact-time +claim locks the pool anchor, an arrival claim locks the window anchor, +so the two capacity transactions never serialize and the ledger has no +row on which to observe that it sold the same staffing twice. +`SchedulingPolicy::check_window_records` is the canonical validation of +the pair and refuses that combination with `shared-supply-unpartitioned`. +Until this change it ran only in `schedulingctl`, against the files on an +operator's disk, so an operator who edited the policy and restarted the +runtime deployed the combination their own tooling refuses. + +**Enforcement:** both database writes re-run that same check under the +locks they already hold. `apply_policy` loads the deployed window records +before it writes a revision, and `replace_facts` loads the retained +document of the deployed policy revision before it swaps the records. +Neither reimplements the rules: they call the authoring implementation +and narrow its findings to the contradictions. Two reasons name an +absence rather than a contradiction and are excluded, because ordinary +operator sequencing depends on both: `unknown-window`, a policy published +before the records that serve it, which every admission on that offering +already refuses loudly (SCHEDULING-SEC-02), and `unknown-offering`, a +deployed window whose offering this publication retires, without which an +arrival offering could never be withdrawn. Either write is refused whole, +naming each contradicting field in the operator's own vocabulary, and +leaves the policy revision, the window records, and the facts revision +unchanged. + +**Tests:** +`records_replacement_refuses_a_window_staffed_by_an_exact_time_pool` and +`policy_publication_refuses_an_exact_time_offering_on_a_deployed_windows_staffing` +(`crates/registry-scheduling/tests/postgres_commitments.rs`) take the +refusal from each direction, and +`an_unpartitioned_shared_staffing_block_is_rejected` +(`crates/registry-scheduling-core/src/policy.rs`) holds the rule they +share. The per-admission refusal remains as defense in depth: +`a_window_record_must_belong_to_the_authorized_offering_and_location` +plants its contradicting record directly in the tables, underneath the +writes that now refuse to produce one. + +## One supply identifier, one kind of supply + +**Threat:** `scheduling_supply` is one flat table keyed by `supply_id`, +and both kinds of supply anchor their capacity transactions on a row in +it. Pool identifiers come from the policy package and window identifiers +come from the environment records, authored separately, so nothing made +the two namespaces disjoint. A collision was not symmetric and neither +half was safe. Publishing a window over a live pool identifier hit the +primary key and aborted the facts replacement with a database error +instead of a refusal. Publishing a pool over a live window identifier was +worse: the insert skipped the conflicting row, so the pool quietly +anchored on a row still marked `window`, and the next records replacement +deleted it along with the rest of that kind. From there `lock_supply` +found fewer rows than it asked for and every commitment against that pool +was refused, on a deployment whose policy looked published and whose +records looked current. + +**Enforcement:** the collision is refused before it can be written, by +name, on both sides. `SchedulingPolicy::check` refuses an arrival +offering whose window identifier an exact-time offering already sells, +which is the check `schedulingctl package` runs with no records in hand, +and `check_window` refuses a published window that takes such an +identifier, which reaches both database writes through the combined check +above. Beneath them the two anchor writes are now one `anchor_supply` +insert that returns the kind the row settled on and refuses when that is +not the kind it asked for, so a standing anchor of the other kind is +named rather than aborting one path and being skipped on the other, and +the standing row keeps its kind. `lock_supply` names the identifiers it +could not find instead of reporting a corrupt ledger, which is what +SCHEDULING-SEC-02 already promised the operator's diagnostics would say. + +**Tests:** `a_supply_identifier_may_not_anchor_both_a_pool_and_a_window` +(`crates/registry-scheduling/tests/postgres_commitments.rs`) publishes a +colliding identifier in both directions and asserts the named refusal and +that the standing anchor keeps its kind, and +`a_pool_and_a_window_may_not_share_one_supply_identifier` +(`crates/registry-scheduling-core/src/policy.rs`) holds the authoring +refusal that reports it offline. + ## Known deferrals -The matrix records five deferrals with their compensating controls. -SCHEDULING-DEF-01 is stated in threat 2 above; the other four are +The matrix records four deferrals with their compensating controls. +SCHEDULING-DEF-01 is stated in threat 2 above; the other three are restated here as the index the matrix's `recordedIn` points at: - **SCHEDULING-DEF-04, the channel is not bound to the verified caller.** @@ -277,23 +358,6 @@ restated here as the index the matrix's `recordedIn` points at: supporting index is partial rather than unique. - **SCHEDULING-DEF-06, retention scope.** Recorded in `RUNTIME-CONFIG.md`, which states plainly what the sweeps cover. -- **SCHEDULING-DEF-07, the combined policy and records invariants are - not re-checked at the database.** `SchedulingPolicy::check_window_records` - validates a policy against the published window records it governs, - and among other rules refuses a window whose staffing pool also backs - an exact-time offering: the two modes count that staffing differently, - and an exact-time claim locks the pool anchor while an arrival claim - locks the window anchor, so the ledger has no row on which to observe - the conflict. That check runs only in `schedulingctl`, against the - policy file on disk. `apply_policy` locks every supply anchor and - fences offerings carrying active claims, but never reads - `scheduling_windows`; `replace_facts` swaps the window records without - reading the deployed policy. An operator who edits the policy and - restarts the runtime can therefore deploy the combination the - authoring check refuses. It is an operator path, not a caller-reachable - one, and the authoring refusal is the compensating control; closing it - means re-checking both directions inside the two transactions that - already hold the anchors locked. A change that closes one of these promotes the matrix entry in the same commit and rewrites this section with it. diff --git a/products/scheduling/contracts/security-invariant-matrix.yaml b/products/scheduling/contracts/security-invariant-matrix.yaml index 0002a6bee..de9721660 100644 --- a/products/scheduling/contracts/security-invariant-matrix.yaml +++ b/products/scheduling/contracts/security-invariant-matrix.yaml @@ -431,6 +431,51 @@ invariants: negativeTest: path: crates/registry-scheduling/tests/postgres_commitments.rs name: records_replacement_refuses_to_move_a_window_with_standing_commitments + - id: SCHEDULING-SEC-26 + state: enforced + threat: >- + A deployment ends up holding a policy and window records that contradict + each other, most consequentially a pool that staffs a published window + while also backing an exact-time offering. The two modes anchor on + different supply rows, so their capacity transactions never serialize + and the ledger cannot observe that it sold the same staffing twice. + enforcementPoint: >- + Both database writes re-run the canonical combined check under the locks + they already hold: policy publication loads the deployed window records + before it writes a revision, and records replacement loads the deployed + policy document before it swaps the records. The two findings that name + an absence rather than a contradiction, a window not published yet and + an offering this publication retires, are excluded so that ordinary + operator sequencing in either order stays possible. + refusal: >- + Refuse the write, naming every contradicting field in the same authoring + vocabulary the offline check uses, and leave the policy revision, the + window records, and the facts revision unchanged. + negativeTest: + path: crates/registry-scheduling/tests/postgres_commitments.rs + name: records_replacement_refuses_a_window_staffed_by_an_exact_time_pool + - id: SCHEDULING-SEC-27 + state: enforced + threat: >- + A resource pool and a published window take the same supply identifier. + Both anchor their capacity transactions on the one row keyed by it, and + the two namespaces are authored separately, so nothing else makes them + disjoint. The pool then serializes against a window's anchor, and the + next records replacement deletes it, leaving every commitment against + that pool refused until the policy is republished. + enforcementPoint: >- + The authoring checks refuse the collision on both sides, against the + policy alone and against the policy with the window records it governs, + so packaging reports it as well. The two anchor writes then share one + insert that returns the kind it settled on, which makes a standing row + of the other kind a named refusal rather than a primary key abort on one + path and a silently skipped anchor on the other. + refusal: >- + Refuse the write, naming the identifier and the kind of supply that + already holds it, and leave the standing anchor and its kind unchanged. + negativeTest: + path: crates/registry-scheduling/tests/postgres_commitments.rs + name: a_supply_identifier_may_not_anchor_both_a_pool_and_a_window deferred: - id: SCHEDULING-DEF-01 subject: Full task-grant bounds re-checked inside the capacity transaction @@ -482,21 +527,3 @@ deferred: The configuration reference says so explicitly rather than implying a sweep that does not run. recordedIn: products/scheduling/RUNTIME-CONFIG.md - - id: SCHEDULING-DEF-07 - subject: Combined policy and records invariants re-checked at the database - state: deferred - reason: >- - The canonical validation of a policy against the published window records - it governs runs only in the authoring tooling, against the policy file on - the operator's disk. Neither database write repeats it: policy - publication fences offerings carrying active claims but never reads the - window records, and records replacement never reads the deployed policy. - An operator who publishes a policy their own tooling would refuse can - therefore back an exact-time offering with a pool that already staffs a - published window, and the two modes then draw on the same staffing - through separate anchor rows the capacity transaction cannot compare. - compensatingControl: >- - Every authored path runs the combined check before either document - reaches a deployment, and names the offending field in the operator's - own vocabulary. Reaching the gap means bypassing that tooling. - recordedIn: products/scheduling/SECURITY-REVIEW-NOTES.md diff --git a/products/scheduling/contracts/security-test-traceability.yaml b/products/scheduling/contracts/security-test-traceability.yaml index 60889c3fd..155505566 100644 --- a/products/scheduling/contracts/security-test-traceability.yaml +++ b/products/scheduling/contracts/security-test-traceability.yaml @@ -75,6 +75,8 @@ entries: - {path: crates/registry-scheduling/tests/postgres_commitments.rs, name: cancellation_respects_the_cutoff_and_replays_its_verdict} - {path: crates/registry-scheduling/tests/postgres_commitments.rs, name: a_concurrent_identical_request_replays_the_winning_receipt} - {path: crates/registry-scheduling/tests/postgres_commitments.rs, name: concurrent_identical_admissible_requests_replay_one_winning_success} + - {path: crates/registry-scheduling/tests/postgres_commitments.rs, name: a_retry_replays_its_receipt_after_its_offering_leaves_the_policy} + - {path: crates/registry-scheduling/tests/postgres_commitments.rs, name: a_retry_replays_its_receipt_after_its_offering_changes_service} - id: SCHEDULING-SEC-10 tests: - {path: crates/registry-scheduling-core/src/admission.rs, name: policy_change_outranks_a_window_revision_mismatch} @@ -184,3 +186,12 @@ entries: - {path: crates/registry-scheduling-core/src/policy.rs, name: moving_a_published_window_reports_the_stranded_commitments} - {path: crates/registry-scheduling/tests/postgres_commitments.rs, name: records_replacement_refuses_to_move_a_window_with_standing_commitments} - {path: crates/registry-scheduling/tests/postgres_commitments.rs, name: policy_publication_refuses_to_remove_an_offering_with_a_live_appointment} + - id: SCHEDULING-SEC-26 + tests: + - {path: crates/registry-scheduling/tests/postgres_commitments.rs, name: records_replacement_refuses_a_window_staffed_by_an_exact_time_pool} + - {path: crates/registry-scheduling/tests/postgres_commitments.rs, name: policy_publication_refuses_an_exact_time_offering_on_a_deployed_windows_staffing} + - {path: crates/registry-scheduling-core/src/policy.rs, name: an_unpartitioned_shared_staffing_block_is_rejected} + - id: SCHEDULING-SEC-27 + tests: + - {path: crates/registry-scheduling/tests/postgres_commitments.rs, name: a_supply_identifier_may_not_anchor_both_a_pool_and_a_window} + - {path: crates/registry-scheduling-core/src/policy.rs, name: a_pool_and_a_window_may_not_share_one_supply_identifier} diff --git a/products/scheduling/generated/registry-scheduling.openapi.json b/products/scheduling/generated/registry-scheduling.openapi.json index b97b4b1d9..e54f42255 100644 --- a/products/scheduling/generated/registry-scheduling.openapi.json +++ b/products/scheduling/generated/registry-scheduling.openapi.json @@ -806,7 +806,7 @@ "const": "booking.duplicate-active" }, "detail": { - "const": "An active booking already holds this party's duplicate key. Cancel or complete it before booking again." + "const": "An active booking already holds this party's duplicate key. Cancel it, or wait until it has passed, before booking again." }, "status": { "const": 409