Skip to content
Merged
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 15 additions & 19 deletions src/postgres_data_rights.rs
Original file line number Diff line number Diff line change
Expand Up @@ -165,8 +165,9 @@ pub fn apply_data_rights_migration(client: &mut Client) -> Result<(), postgres::
/// The function owns one short local `PostgreSQL` transaction so a database or outbox error rolls
/// back the request row, target rows, and any earlier event inserts together. It does not deliver
/// events itself; the existing outbox worker owns retry, quarantine, and reconciliation behavior.
/// Exact request replay is idempotent only when the target set and outbox evidence are unchanged.
/// The insert-then-inspect first-write classifier requires `READ COMMITTED`, matching the existing
/// Exact creation replay remains idempotent after the stored lifecycle advances when the immutable
/// request evidence, target set, event identities, and outbox evidence are unchanged. The
/// insert-then-inspect first-write classifier requires `READ COMMITTED`, matching the existing
/// integration outbox replay contract, and fails closed when the session default uses stronger
/// isolation.
///
Expand Down Expand Up @@ -222,11 +223,11 @@ pub fn persist_requested_data_rights_with_propagation(

/// Persist requester identity verification for one already requested data-rights identity.
///
/// Exact replay of the same evidence and verification time is idempotent. A later
/// conflicting verification fails closed. Replay classification locks the matched
/// request row until the caller-owned transaction ends so the classified lifecycle
/// cannot change before the caller composes subsequent atomic work.
/// This adapter does not start processing or complete the request.
/// Exact replay of the same evidence and verification time is idempotent, including after later
/// lifecycle transitions that preserve the original verification evidence. A conflicting
/// verification fails closed. Replay classification locks the matched request row until the
/// caller-owned transaction ends so the classified lifecycle cannot change before the caller
/// composes subsequent atomic work. This adapter does not start processing or complete the request.
///
/// # Errors
///
Expand Down Expand Up @@ -287,7 +288,7 @@ pub fn persist_data_rights_identity_verification(
let row = query_optional_row(
transaction,
"SELECT participant_ref, request_kind, scope_ref,
current_state, verification_evidence_ref, verified_at_unix_ms
verification_evidence_ref, verified_at_unix_ms
FROM data_rights_request_state
WHERE request_ref = $1 AND tenant_ref = $2
FOR UPDATE",
Expand All @@ -299,20 +300,18 @@ pub fn persist_data_rights_identity_verification(
let stored_participant: String = row.get(0);
let stored_kind: String = row.get(1);
let stored_scope: String = row.get(2);
let stored_state: String = row.get(3);
let stored_evidence: Option<String> = row.get(4);
let stored_verified_at: Option<i64> = row.get(5);
let stored_evidence: Option<String> = row.get(3);
let stored_verified_at: Option<i64> = row.get(4);
let identity_matches = stored_participant == request.participant_ref()
&& stored_kind == request_kind
&& stored_scope == request.scope_ref();
if !identity_matches {
Err(DataRightsPersistenceError::ConflictingReplay)
} else if stored_state == "identity_verified"
&& stored_evidence.as_deref() == Some(evidence_ref)
} else if stored_evidence.as_deref() == Some(evidence_ref)
&& stored_verified_at == Some(verified_at)
{
Ok(DataRightsVerificationDisposition::Duplicate)
} else if stored_state == "identity_verified" {
} else if stored_evidence.is_some() || stored_verified_at.is_some() {
Err(DataRightsPersistenceError::ConflictingReplay)
} else {
Err(DataRightsPersistenceError::InvalidRequestState)
Expand Down Expand Up @@ -355,18 +354,15 @@ fn persist_request_header(
}

let row = transaction.query_one(
"SELECT tenant_ref, participant_ref, request_kind, scope_ref, current_state, \
requested_at_unix_ms, latest_event_at_unix_ms \
"SELECT tenant_ref, participant_ref, request_kind, scope_ref, requested_at_unix_ms \
FROM data_rights_request_state WHERE request_ref = $1",
&[&request.request_ref()],
)?;
let exact = row.get::<_, String>(0) == request.tenant_ref()
&& row.get::<_, String>(1) == request.participant_ref()
&& row.get::<_, String>(2) == request_kind_name(request.kind())
&& row.get::<_, String>(3) == request.scope_ref()
&& row.get::<_, String>(4) == "requested"
&& row.get::<_, i64>(5) == requested_at
&& row.get::<_, i64>(6) == requested_at;
&& row.get::<_, i64>(4) == requested_at;
if exact && stored_targets_match(transaction, request, targets)? {
Ok(DataRightsPersistenceDisposition::Duplicate)
} else {
Expand Down
134 changes: 134 additions & 0 deletions tests/postgres_data_rights_creation_replay.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,134 @@
//! Creation and verification replay stay idempotent after the durable request lifecycle advances.

use postgres::{Client, NoTls};
use psychometrics_commons_runtime::data_rights::{DataRightsRequest, DataRightsRequestKind};
use psychometrics_commons_runtime::integration::IntegrationEvent;
use psychometrics_commons_runtime::postgres_data_rights::{
apply_data_rights_migration, persist_data_rights_identity_verification,
persist_requested_data_rights_with_propagation, DataRightsPersistenceDisposition,
DataRightsPropagationTarget, DataRightsVerificationDisposition,
};
use psychometrics_commons_runtime::postgres_integration::apply_integration_migration;

const DIGEST: &str = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";

fn ready_client() -> Client {
let url = std::env::var("TEST_DATABASE_URL").expect("TEST_DATABASE_URL is required");
let mut client = Client::connect(&url, NoTls).expect("CI PostgreSQL must be reachable");
let schema = format!("data_rights_creation_replay_{}", std::process::id());
client
.batch_execute(&format!(
"DROP SCHEMA IF EXISTS {schema} CASCADE; CREATE SCHEMA {schema}; SET search_path TO {schema};"
))
.unwrap();
Comment thread
github-actions[bot] marked this conversation as resolved.
Resolved
apply_integration_migration(&mut client).unwrap();
apply_data_rights_migration(&mut client).unwrap();
client
}

fn request() -> DataRightsRequest {
DataRightsRequest::new(
"data_rights_request_creation_replay",
"tenant_alpha",
"participant_alpha",
DataRightsRequestKind::Deletion,
"scope_alpha",
10_000,
)
.unwrap()
}

fn event() -> IntegrationEvent {
IntegrationEvent::new(
"data_rights_event_creation_replay",
"data_rights.deletion.requested",
"v1",
"psychometrics_commons",
"tenant_alpha",
"data_rights_request_creation_replay",
10_000,
"data_rights_request_creation_replay",
None,
DIGEST,
)
.unwrap()
}

fn persist_requested(client: &mut Client) {
let event = event();
let targets = [DataRightsPropagationTarget::new(
"dependent_system_alpha",
&event,
)];
assert_eq!(
persist_requested_data_rights_with_propagation(client, &request(), &targets, 3).unwrap(),
DataRightsPersistenceDisposition::Inserted
);
}

fn verified_request() -> DataRightsRequest {
let mut request = request();
request
.verify_identity("verification_evidence_creation_replay", 10_100)
.unwrap();
request
}

#[test]
fn exact_creation_replay_remains_duplicate_after_identity_verification() {
let mut client = ready_client();
let event = event();
let targets = [DataRightsPropagationTarget::new(
"dependent_system_alpha",
&event,
)];

persist_requested(&mut client);

let advanced = verified_request();
{
let mut transaction = client.transaction().unwrap();
persist_data_rights_identity_verification(&mut transaction, &advanced).unwrap();
transaction.commit().unwrap();
}

assert_eq!(
persist_requested_data_rights_with_propagation(&mut client, &request(), &targets, 3)
.unwrap(),
DataRightsPersistenceDisposition::Duplicate,
"retrying immutable creation evidence must not conflict merely because the stored lifecycle advanced"
);
}

#[test]
fn exact_identity_verification_replay_remains_duplicate_after_processing_state() {
let mut client = ready_client();
persist_requested(&mut client);

let verified = verified_request();
{
let mut transaction = client.transaction().unwrap();
assert_eq!(
persist_data_rights_identity_verification(&mut transaction, &verified).unwrap(),
DataRightsVerificationDisposition::Verified
);
transaction.commit().unwrap();
}

client
.execute(
"UPDATE data_rights_request_state
SET current_state = 'processing', latest_event_at_unix_ms = $2
WHERE request_ref = $1",
&[&verified.request_ref(), &10_200_i64],
)
.unwrap();

let mut transaction = client.transaction().unwrap();
assert_eq!(
persist_data_rights_identity_verification(&mut transaction, &verified).unwrap(),
DataRightsVerificationDisposition::Duplicate,
"retrying immutable verification evidence must remain idempotent after a later lifecycle transition"
);
transaction.commit().unwrap();
}
Loading