feat(sync): rfc-003 phase a.2.1 — dual-shape ingest on /sync/ops

src/api/sync.rs

@@ -40,7 +40,7 @@ use uuid::Uuid;
 use crate::{
     db,
     middleware::UserId,
-    sync::{build_broadcast, SyncOp, SyncOpIn},
+    sync::{build_broadcast, Hlc, SyncOp, SyncOpIn},
     AppState,
 };
 
@@ -79,6 +79,19 @@ pub struct LamportRegression {
     pub offending_lamport_ts: i64,
 }
 
+/// 409 body when a v2 client pushes an `hlc` pair that collides on the
+/// `(user_id, device_id, hlc_wall, hlc_logical)` UNIQUE — RFC-003 §2
+/// requires per-device HLC monotonicity, so this is the v2 equivalent
+/// of the lamport-regression case. The offending pair is echoed so
+/// the client can resync its HLC instead of guessing how far the
+/// server has advanced.
+#[derive(Debug, Serialize, ToSchema)]
+pub struct HlcRegression {
+    pub error: &'static str,
+    pub device_id: String,
+    pub offending_hlc: Hlc,
+}
+
 #[derive(Debug, Deserialize, ToSchema, utoipa::IntoParams)]
 pub struct PullQuery {
     /// Last `sync_op.id` the client has confirmed seeing. `0` (or
@@ -140,9 +153,17 @@ pub fn router(state: AppState) -> OpenApiRouter {
     request_body = PushBatchRequest,
     responses(
         (status = 200, description = "Batch accepted (one entry per op, fresh or dup)", body = PushBatchResponse),
-        (status = 400, description = "Empty `device_id`, oversized batch, or malformed op"),
+        (status = 400, description = "Empty `device_id`, oversized batch, malformed op, or hlc.wall/logical < 0"),
         (status = 401, description = "Missing or invalid bearer token"),
-        (status = 409, description = "Lamport regression — stored max returned in body", body = LamportRegression),
+        // Two regression shapes can land on 409 — discriminated on the
+        // `error` field of the body. The legacy v1 path returns
+        // `LamportRegression { error: \"lamport_regression\", stored_max, offending_lamport_ts }`;
+        // the v2 path returns `HlcRegression { error: \"hlc_regression\", offending_hlc }`.
+        // utoipa 5 has no concise `oneOf` for response bodies, so we
+        // list both shapes as sibling 409 entries — clients pattern-
+        // match on `error` before reading the discriminating fields.
+        (status = 409, description = "Lamport regression (v1) — body discriminated by `error: \"lamport_regression\"`", body = LamportRegression),
+        (status = 409, description = "HLC collision (v2) — body discriminated by `error: \"hlc_regression\"`", body = HlcRegression),
         (status = 500, description = "Database or internal failure"),
     ),
 )]
@@ -193,6 +214,31 @@ async fn push_ops(
                 .into_response();
         }
 
+        // RFC-003 Phase A.2 — v2 wire shape carries an explicit
+        // `hlc` pair. Validate `wall >= 0` (logical is already i32
+        // by the type, so the only out-of-range a v2 client can hit
+        // is a negative wall — usually a clock-set bug). The §2
+        // tiebreaker `origin_device_id` rides through the existing
+        // `device_id` string per A.1.1's design. The server never
+        // tries to "fix up" a missing hlc by synthesising one — the
+        // v1 path's `(0, lamport_ts)` derivation owns that case.
+        if let Some(hlc) = op_in.hlc {
+            if hlc.wall < 0 {
+                tx.rollback().await.ok();
+                return (StatusCode::BAD_REQUEST, "hlc.wall must be >= 0").into_response();
+            }
+            // Symmetric with `wall`. `logical` is `i32` by the type so
+            // a negative is structurally legal but semantically wrong
+            // per RFC-003 §2 (unsigned-shaped counter). Catching it
+            // here returns 400 instead of letting the db helper's
+            // defence-in-depth guard surface as a 500.
+            if hlc.logical < 0 {
+                tx.rollback().await.ok();
+                return (StatusCode::BAD_REQUEST, "hlc.logical must be >= 0").into_response();
+            }
+        }
+        let hlc_pair = op_in.hlc.map(|h| (h.wall, h.logical));
+
         let insert_res = db::sync::insert_op_returning(
             &mut tx,
             user_id,
@@ -206,6 +252,7 @@ async fn push_ops(
             op_in.payload.as_ref(),
             now,
             op_in.profile_canonical_id.as_deref(),
+            hlc_pair,
         )
         .await;
 
@@ -253,33 +300,79 @@ async fn push_ops(
                 }
             }
             Err(sqlx::Error::Database(db_err)) if db_err.code().as_deref() == Some("23505") => {
-                // Unique violation that wasn't on `operation_id`
-                // (that branch is absorbed by the ON CONFLICT) — so it
-                // can only be the `(user_id, device_id, lamport_ts)`
-                // constraint, i.e. a regression. Read the current max
-                // and return it so the client can resync its clock.
+                // After A.1.1 two UNIQUE constraints can fire 23505 on
+                // a regression. The third (`operation_id`) is absorbed
+                // upstream by `ON CONFLICT DO NOTHING`, so we never see
+                // it here. Discriminate on the constraint name so a v2
+                // HLC collision doesn't get reported as a misleading
+                // lamport regression (the stored lamport_max would be
+                // meaningless to a v2 client whose clock is the HLC
+                // pair, not lamport_ts).
                 tx.rollback().await.ok();
-                let stored_max = match db::sync::lamport_max(pool, user_id, device_id).await {
-                    Ok(n) => n,
-                    Err(err) => {
-                        // Surface the read failure rather than masking
-                        // it behind a `0` that would tell the client
-                        // "your clock is fine, retry" when it isn't.
-                        tracing::error!(error = %err, user_id, device_id, "lamport_max read failed");
-                        return (StatusCode::INTERNAL_SERVER_ERROR, "lamport_max read failed")
+                match db_err.constraint() {
+                    Some("sync_op_user_device_hlc_uniq") => {
+                        // V2 client pushed an HLC pair already taken
+                        // by this device. Echo the offending pair so
+                        // the client can resync; `op_in.hlc` is
+                        // guaranteed `Some` here because the v1 path
+                        // derives `(0, lamport_ts)` from a strictly-
+                        // increasing lamport_ts (per the legacy
+                        // `(user_id, device_id, lamport_ts)` UNIQUE
+                        // also fires on regression), so a v1 client
+                        // hitting THIS constraint exclusively would
+                        // mean lamport_ts moved forward but the
+                        // derived `(0, lamport_ts)` collided — only
+                        // possible after a manual DB reset, which is
+                        // out of scope. Default the pair anyway in
+                        // case a future code path bypasses the v2
+                        // gate.
+                        return (
+                            StatusCode::CONFLICT,
+                            Json(HlcRegression {
+                                error: "hlc_regression",
+                                device_id: device_id.to_string(),
+                                offending_hlc: op_in.hlc.unwrap_or(Hlc {
+                                    wall: 0,
+                                    logical: 0,
+                                }),
+                            }),
+                        )
                             .into_response();
                     }
-                };
-                return (
-                    StatusCode::CONFLICT,
-                    Json(LamportRegression {
-                        error: "lamport_regression",
-                        device_id: device_id.to_string(),
-                        stored_max,
-                        offending_lamport_ts: op_in.lamport_ts,
-                    }),
-                )
-                    .into_response();
+                    _ => {
+                        // Legacy `(user_id, device_id, lamport_ts)`
+                        // constraint (auto-named by Postgres). Read
+                        // the current max and return it so the v1
+                        // client can resync its clock.
+                        let stored_max =
+                            match db::sync::lamport_max(pool, user_id, device_id).await {
+                                Ok(n) => n,
+                                Err(err) => {
+                                    // Surface the read failure rather
+                                    // than masking it behind a `0`
+                                    // that would tell the client
+                                    // "your clock is fine, retry"
+                                    // when it isn't.
+                                    tracing::error!(error = %err, user_id, device_id, "lamport_max read failed");
+                                    return (
+                                        StatusCode::INTERNAL_SERVER_ERROR,
+                                        "lamport_max read failed",
+                                    )
+                                        .into_response();
+                                }
+                            };
+                        return (
+                            StatusCode::CONFLICT,
+                            Json(LamportRegression {
+                                error: "lamport_regression",
+                                device_id: device_id.to_string(),
+                                stored_max,
+                                offending_lamport_ts: op_in.lamport_ts,
+                            }),
+                        )
+                            .into_response();
+                    }
+                }
             }
             Err(err) => {
                 tracing::error!(error = %err, user_id, device_id, "sync insert failed");
@@ -533,5 +626,9 @@ fn row_to_op(row: &PgRow) -> SyncOp {
         payload: row.get::<Option<serde_json::Value>, _>("payload"),
         created_at: row.get("created_at"),
         profile_canonical_id: row.get("profile_canonical_id"),
+        hlc: Hlc {
+            wall: row.get("hlc_wall"),
+            logical: row.get("hlc_logical"),
+        },
     }
 }

src/db.rs

@@ -93,6 +93,15 @@ pub mod sync {
     /// UNIQUE is *not* covered by `ON CONFLICT` — a violation there
     /// bubbles up as a `sqlx::Error::Database` with SQLSTATE 23505
     /// for the caller to map to its 409 path.
+    ///
+    /// RFC-003 Phase A.2 — `hlc` is the v2 wire-shape addition. When
+    /// `hlc.is_none()`, the legacy A.1.1 derivation `(0, lamport_ts)`
+    /// kicks in so v1 clients keep working unchanged. When
+    /// `hlc.is_some()`, the caller's pair is bound verbatim — the
+    /// caller validates `hlc.wall >= 0` before reaching this helper
+    /// (`logical` is already i32 by the type). The §2 tiebreaker
+    /// `origin_device_id` rides through the existing `device_id`
+    /// TEXT column per A.1.1's header rationale.
     #[allow(clippy::too_many_arguments)]
     pub async fn insert_op_returning(
         conn: &mut PgConnection,
@@ -107,41 +116,62 @@ pub mod sync {
         payload: Option<&serde_json::Value>,
         created_at: i64,
         profile_canonical_id: Option<&str>,
+        hlc: Option<(i64, i32)>,
     ) -> Result<Option<PgRow>, sqlx::Error> {
-        // Phase A.1 (RFC-003): every row also carries the HLC pair the
-        // §2 total order is defined on. Until A.2 lands the wire shape
-        // change that lets clients send their own `hlc`, we derive it
-        // from `lamport_ts` exactly the way the 20260612000000 backfill
-        // does — `(0, lamport_ts)`. That keeps the new
+        // V1 path derives the HLC pair from `lamport_ts` exactly the
+        // way the 20260612000000 backfill does — `(0, lamport_ts)`.
+        // V2 path binds the caller's pair verbatim. Either way the
         // `UNIQUE (user_id, device_id, hlc_wall, hlc_logical)` invariant
-        // satisfied without touching callers, and means a v2 op
-        // (`hlc_wall > 0`) strictly outranks every legacy-shape row
-        // under the §2 total order once A.2 ships.
+        // is satisfied: per-device pairs are monotonic by construction
+        // (lamport_ts strictly increasing → derived `(0, lamport_ts)`
+        // strictly increasing; v2 clients enforce per-device HLC
+        // monotonicity client-side per RFC-003 §2).
         //
-        // `hlc_logical` is INTEGER (i32) per the RFC §2 definition of
-        // the logical counter. Validate the incoming `lamport_ts`
-        // before binding so a hypothetical >2^31 value surfaces as a
-        // typed error instead of Postgres's bare "integer out of
-        // range" SQLSTATE 22003 — A.2's dedicated v2 column gains its
-        // own narrower binding, but until then the legacy path needs
-        // the gate.
-        if !(0..=i64::from(i32::MAX)).contains(&lamport_ts) {
-            return Err(sqlx::Error::Protocol(format!(
-                "lamport_ts {lamport_ts} is out of range for hlc_logical (i32); widen the column or reset the device counter"
-            )));
-        }
-        let hlc_logical: i32 = lamport_ts as i32;
+        // `hlc_logical` is INTEGER (i32) per the RFC §2 definition. On
+        // the v1 path we still validate `lamport_ts` before narrowing —
+        // a hypothetical >2^31 value would silently truncate otherwise.
+        let (hlc_wall, hlc_logical) = match hlc {
+            Some((wall, logical)) => {
+                // V2 path defence in depth — the API boundary already
+                // rejects `wall < 0`, but the helper is shared so a
+                // future caller bypassing the handler still gets a
+                // typed error instead of a row with an invalid §2
+                // total-order tuple. `logical < 0` cannot reach here
+                // through the wire shape (the type is i32 so a
+                // negative is structurally legal but semantically
+                // wrong — RFC-003 §2 defines the logical counter as
+                // u32) but is rejected for the same total-order
+                // invariant. Mirrors the v1 path's `Protocol` error
+                // shape so the push handler maps both to a 500 +
+                // structured log.
+                if wall < 0 || logical < 0 {
+                    return Err(sqlx::Error::Protocol(format!(
+                        "hlc ({wall}, {logical}) is out of range for the §2 total order (both components must be >= 0)"
+                    )));
+                }
+                (wall, logical)
+            }
+            None => {
+                if !(0..=i64::from(i32::MAX)).contains(&lamport_ts) {
+                    return Err(sqlx::Error::Protocol(format!(
+                        "lamport_ts {lamport_ts} is out of range for hlc_logical (i32); widen the column or reset the device counter"
+                    )));
+                }
+                (0i64, lamport_ts as i32)
+            }
+        };
         sqlx::query(
             "INSERT INTO sync_op \
                 (user_id, device_id, operation_id, lamport_ts, hlc_wall, hlc_logical, entity, entity_id, field, op, payload, created_at, profile_canonical_id) \
-             VALUES ($1, $2, $3, $4, 0, $5, $6, $7, $8, $9, $10, $11, $12) \
+             VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) \
              ON CONFLICT (user_id, device_id, operation_id) DO NOTHING \
-             RETURNING id, operation_id, device_id, lamport_ts, entity, entity_id, field, op, payload, created_at, profile_canonical_id",
+             RETURNING id, operation_id, device_id, lamport_ts, hlc_wall, hlc_logical, entity, entity_id, field, op, payload, created_at, profile_canonical_id",
         )
         .bind(user_id)
         .bind(device_id)
         .bind(operation_id)
         .bind(lamport_ts)
+        .bind(hlc_wall)
         .bind(hlc_logical)
         .bind(entity)
         .bind(entity_id)
@@ -165,7 +195,7 @@ pub mod sync {
         operation_id: Uuid,
     ) -> Result<PgRow, sqlx::Error> {
         sqlx::query(
-            "SELECT id, operation_id, device_id, lamport_ts, entity, entity_id, field, op, payload, created_at, profile_canonical_id \
+            "SELECT id, operation_id, device_id, lamport_ts, hlc_wall, hlc_logical, entity, entity_id, field, op, payload, created_at, profile_canonical_id \
              FROM sync_op \
              WHERE user_id = $1 AND device_id = $2 AND operation_id = $3",
         )
@@ -222,7 +252,7 @@ pub mod sync {
         limit: i64,
     ) -> Result<Vec<PgRow>, sqlx::Error> {
         sqlx::query(
-            "SELECT id, operation_id, device_id, lamport_ts, entity, entity_id, field, op, payload, created_at, profile_canonical_id \
+            "SELECT id, operation_id, device_id, lamport_ts, hlc_wall, hlc_logical, entity, entity_id, field, op, payload, created_at, profile_canonical_id \
              FROM sync_op \
              WHERE user_id = $1 AND id > $2 \
              ORDER BY id ASC \

src/sync.rs

@@ -76,6 +76,25 @@ pub const DEFAULT_COMPACTION_INTERVAL: Duration = Duration::from_secs(24 * 60 *
 /// permanently-lost device doesn't drag the log forever.
 pub const STALE_DEVICE_MS: i64 = 90 * 24 * 60 * 60 * 1000;
 
+/// Hybrid Logical Clock pair carried by RFC-003 v2 ops on the wire.
+///
+/// `wall` is epoch-millis (BIGINT in Postgres). `logical` is the
+/// per-tick counter the HLC paper defines as `u32` — Postgres stores
+/// it as INTEGER (i32), and `src/db.rs::insert_op_returning` enforces
+/// `0..=i32::MAX` on bind. The narrowing is documented in the A.1.1
+/// migration header alongside the escalation path.
+///
+/// Carried on the v2 wire shape alongside `origin_device_id`; the
+/// (`hlc`, `origin_device_id`) tuple is the §2 total order the apply
+/// pipeline LWW reasoning runs on. Phase A.2 only ingests + echoes
+/// the pair; the apply-side propagation onto entity rows lands in
+/// Phase A.2.2.
+#[derive(Debug, Clone, Copy, Deserialize, Serialize, ToSchema, PartialEq, Eq)]
+pub struct Hlc {
+    pub wall: i64,
+    pub logical: i32,
+}
+
 /// Wire format for a single op the client pushes. `payload` stays
 /// opaque to the server — it's stored as JSONB and replayed verbatim
 /// to other devices, so the schema can evolve client-side without a
@@ -108,6 +127,22 @@ pub struct SyncOpIn {
     /// ops still land in the durable log but skip the apply path.
     #[serde(default)]
     pub profile_canonical_id: Option<String>,
+    /// RFC-003 Phase A.2 — Hybrid Logical Clock carried by v2 desktop
+    /// clients. `None` from v1 clients; the server derives the pair
+    /// from `lamport_ts` (wall = 0, logical = lamport_ts) when this is
+    /// absent, exactly the way the A.1.1 backfill did. A v2 op with
+    /// `hlc.wall > 0` strictly outranks every v1-derived row under
+    /// the §2 total order.
+    ///
+    /// The §2 tiebreaker `origin_device_id` rides through the
+    /// existing [`PushBatchRequest::device_id`] string (per A.1.1's
+    /// header — UUID-shaped TEXT round-trips without loss). v2
+    /// clients format their `device_id` as a UUID; v1 clients keep
+    /// their free-form string. Either way the wire shape stays a
+    /// single string, and the apply pipeline (A.2.2) parses it as
+    /// UUID when writing `origin_device_id` onto the entity row.
+    #[serde(default)]
+    pub hlc: Option<Hlc>,
 }
 
 /// Wire format for an accepted op. Mirrors the row shape so the
@@ -134,6 +169,11 @@ pub struct SyncOp {
     /// so pulling devices can land the op in the right profile.
     #[serde(default)]
     pub profile_canonical_id: Option<String>,
+    /// RFC-003 Phase A.2 echo. Always populated on read — the server
+    /// stamps `(0, lamport_ts)` for v1-shape rows so a pulling v2
+    /// client sees a usable (if minimal) total-order tuple. v2-shape
+    /// rows echo the originator's pair verbatim.
+    pub hlc: Hlc,
 }
 
 /// Broadcast payload. We pre-serialise to JSON once at emit time so