Skip to content
23 changes: 19 additions & 4 deletions src/longitudinal_observation.rs
Original file line number Diff line number Diff line change
Expand Up @@ -233,6 +233,8 @@ pub enum LongitudinalObservationError {
NonMonotonicIngestion,
/// The same source observation was replayed with different evidence.
IdempotencyConflict,
/// A distinct source observation tried to reuse an existing Commons record identity.
ObservationIdentityConflict,
}

impl Display for LongitudinalObservationError {
Expand Down Expand Up @@ -274,6 +276,9 @@ impl Display for LongitudinalObservationError {
Self::IdempotencyConflict => {
"replay the first accepted observation for this source identity; do not last-write-win a new clock"
}
Self::ObservationIdentityConflict => {
"mint a new Commons observation-record identity for a distinct source observation; do not reuse an existing record identity"
}
})
}
}
Expand Down Expand Up @@ -311,14 +316,17 @@ impl LongitudinalObservationSet {
///
/// Source identity is `(enrollment_ref, source_system_ref, source_observation_ref)`.
/// An exact retry is a no-op. A retry with different clocks, membership, or
/// construct evidence fails closed. Source clocks that arrive after receipt are
/// flagged and kept. Platform ingest cannot precede receipt or the previous
/// accepted ingest time.
/// construct evidence fails closed. A distinct source observation must also
/// receive a distinct Commons `observation_record_ref`; a reused record identity
/// fails closed instead of aliasing two immutable observations. Source clocks
/// that arrive after receipt are flagged and kept. Platform ingest cannot precede
/// receipt or the previous accepted ingest time.
///
/// # Errors
///
/// Returns [`LongitudinalObservationError`] when a reference, clock, timezone,
/// membership share, platform order, or source-identity replay is invalid.
/// membership share, platform order, source-identity replay, or Commons record
/// identity reuse is invalid.
pub fn ingest(
&mut self,
input: LongitudinalObservationInput<'_>,
Expand All @@ -334,6 +342,13 @@ impl LongitudinalObservationSet {
}
return Err(LongitudinalObservationError::IdempotencyConflict);
}
if self
.records
.iter()
.any(|record| record.observation_record_ref == candidate.observation_record_ref)
{
return Err(LongitudinalObservationError::ObservationIdentityConflict);
}
if self
.records
.last()
Expand Down
68 changes: 68 additions & 0 deletions tests/longitudinal_observation_record_identity.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
//! Regression contract for unique Commons longitudinal observation identities.

use psychometrics_commons_runtime::longitudinal_observation::{
LongitudinalObservationError, LongitudinalObservationInput, LongitudinalObservationSet,
MembershipShareInput, ObservationTimeInput,
};

static MEMBERSHIPS: [MembershipShareInput<'static>; 1] = [MembershipShareInput {
membership_context_ref: "seoul_clinic_ward_a",
weight_parts_per_10_000: 10_000,
}];

fn observation<'a>(
observation_record_ref: &'a str,
source_observation_ref: &'a str,
received_at_unix_ms: u64,
ingested_at_unix_ms: u64,
) -> LongitudinalObservationInput<'a> {
LongitudinalObservationInput {
observation_record_ref,
enrollment_ref: "enrollment_seoul_clinic_2026q3",
source_system_ref: "gyeot_ema_android",
source_observation_ref,
construct_ref: "construct_anxious_mood",
measure_ref: "measure_anxious_mood_ema_v3",
membership_shares: &MEMBERSHIPS,
time: ObservationTimeInput {
validity_start_at_unix_ms: 1_776_297_000_000,
validity_end_at_unix_ms: 1_776_297_000_000,
recorded_at_unix_ms: received_at_unix_ms - 1_000,
received_at_unix_ms,
ingested_at_unix_ms,
timezone_name: "Asia/Seoul",
utc_offset_minutes: 540,
},
}
}

#[test]
fn distinct_source_observations_cannot_reuse_one_commons_record_identity() {
let mut observations = LongitudinalObservationSet::new();
observations
.ingest(observation(
"observation_record_shared_identity",
"gyeot_obs_first",
1_776_336_120_000,
1_776_336_180_000,
))
.expect("first source observation owns the Commons record identity");

let conflict = observations
.ingest(observation(
"observation_record_shared_identity",
"gyeot_obs_second",
1_776_336_240_000,
1_776_336_300_000,
))
.expect_err("a distinct source observation must not reuse the Commons record identity");

assert_eq!(
conflict,
LongitudinalObservationError::ObservationIdentityConflict
);
assert!(conflict
.to_string()
.contains("observation-record identity"));
assert_eq!(observations.len(), 1);
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
}
Loading