refactor(core): deepen domain stores and invitation-only persistence

Peel relationships, eligibility, and secrets off the shared pool into
AppDataStores adapters, split pairing service/protocol, and move the
invitation Repository into its own module so open_all owns schemas.

Co-authored-by: Cursor <cursoragent@cursor.com>

crates/vnidrop/src/device_relationship/lifecycle.rs

@@ -1,28 +1,13 @@
 //! Forget, block, grant rotation, and minimal revocation tombstones (design §7–§8).
 
 use serde_json::json;
-use sqlx::Row;
 
-use super::{DeviceRelationshipService, RelationshipRow};
+use super::{store::RelationshipRow, DeviceRelationshipService};
 use crate::{
     api::DeviceRelationshipState, error::VnidropError, grant::GrantRejection,
-    secure_secret::SecretHandle, util::now_ms,
+    secure_secret::SecretHandle,
 };
 
-/// Minimal non-secret tombstone for a revoked relationship generation.
-///
-/// Retains only what is needed to reject replay: peer identity, generation,
-/// opaque grant ids, and revocation time. No names, filenames, history, or
-/// capability material.
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub(crate) struct GenerationTombstone {
-    pub(crate) remote_endpoint_id: String,
-    pub(crate) generation: u64,
-    pub(crate) issued_grant_id: Option<String>,
-    pub(crate) held_grant_id: Option<String>,
-    pub(crate) revoked_at: i64,
-}
-
 #[derive(Debug, Clone)]
 pub(crate) struct ForgetOutcome {
     pub(crate) had_relationship: bool,
@@ -31,24 +16,6 @@ pub(crate) struct ForgetOutcome {
 }
 
 impl DeviceRelationshipService {
-    pub(crate) async fn ensure_lifecycle_schema(pool: &sqlx::SqlitePool) -> anyhow::Result<()> {
-        sqlx::query(
-            r#"
-            CREATE TABLE IF NOT EXISTS relationship_generation_tombstones (
-                remote_endpoint_id TEXT NOT NULL,
-                generation INTEGER NOT NULL,
-                issued_grant_id TEXT,
-                held_grant_id TEXT,
-                revoked_at INTEGER NOT NULL,
-                PRIMARY KEY (remote_endpoint_id, generation)
-            );
-            "#,
-        )
-        .execute(pool)
-        .await?;
-        Ok(())
-    }
-
     /// Forget a saved (or pending) device: revoke locally first, clean secrets,
     /// then the caller sends a best-effort remote notice. Invitation-domain
     /// transfers are untouched.
@@ -127,25 +94,9 @@ impl DeviceRelationshipService {
         self.clear_grant_secrets(&row).await?;
 
         let new_generation = row.generation.saturating_add(1);
-        let now = now_ms();
-        sqlx::query(
-            r#"
-            UPDATE device_relationships
-            SET generation = ?2,
-                issued_grant_handle = NULL,
-                held_grant_handle = NULL,
-                issued_grant_id = NULL,
-                held_grant_id = NULL,
-                updated_at = ?3
-            WHERE remote_endpoint_id = ?1
-            "#,
-        )
-        .bind(&peer_endpoint_id)
-        .bind(new_generation as i64)
-        .bind(now)
-        .execute(&self.pool)
-        .await
-        .map_err(VnidropError::repository)?;
+        self.store
+            .begin_grant_rotation(&peer_endpoint_id, new_generation)
+            .await?;
 
         let _wire = self
             .mint_and_store_issued_grant(
@@ -210,29 +161,8 @@ impl DeviceRelationshipService {
     pub(crate) async fn list_tombstones(
         &self,
         peer_endpoint_id: &str,
-    ) -> Result<Vec<GenerationTombstone>, VnidropError> {
-        let rows = sqlx::query(
-            r#"
-            SELECT remote_endpoint_id, generation, issued_grant_id, held_grant_id, revoked_at
-            FROM relationship_generation_tombstones
-            WHERE remote_endpoint_id = ?1
-            ORDER BY generation ASC
-            "#,
-        )
-        .bind(peer_endpoint_id)
-        .fetch_all(&self.pool)
-        .await
-        .map_err(VnidropError::repository)?;
-        Ok(rows
-            .into_iter()
-            .map(|row| GenerationTombstone {
-                remote_endpoint_id: row.get("remote_endpoint_id"),
-                generation: row.get::<i64, _>("generation") as u64,
-                issued_grant_id: row.get("issued_grant_id"),
-                held_grant_id: row.get("held_grant_id"),
-                revoked_at: row.get("revoked_at"),
-            })
-            .collect())
+    ) -> Result<Vec<super::store::GenerationTombstone>, VnidropError> {
+        self.store.list_tombstones(peer_endpoint_id).await
     }
 
     #[cfg(test)]
@@ -254,52 +184,17 @@ impl DeviceRelationshipService {
         peer_endpoint_id: &str,
         row: &RelationshipRow,
     ) -> Result<(), VnidropError> {
-        sqlx::query(
-            r#"
-            INSERT INTO relationship_generation_tombstones (
-                remote_endpoint_id, generation, issued_grant_id, held_grant_id, revoked_at
-            ) VALUES (?1, ?2, ?3, ?4, ?5)
-            ON CONFLICT(remote_endpoint_id, generation) DO UPDATE SET
-                issued_grant_id = COALESCE(excluded.issued_grant_id, relationship_generation_tombstones.issued_grant_id),
-                held_grant_id = COALESCE(excluded.held_grant_id, relationship_generation_tombstones.held_grant_id),
-                revoked_at = excluded.revoked_at
-            "#,
-        )
-        .bind(peer_endpoint_id)
-        .bind(row.generation as i64)
-        .bind(row.issued_grant_id.as_deref())
-        .bind(row.held_grant_id.as_deref())
-        .bind(now_ms())
-        .execute(&self.pool)
-        .await
-        .map_err(VnidropError::repository)?;
-        Ok(())
+        self.store.insert_tombstone(peer_endpoint_id, row).await
     }
 
     async fn find_tombstone(
         &self,
         peer_endpoint_id: &str,
         generation: u64,
-    ) -> Result<Option<GenerationTombstone>, VnidropError> {
-        let row = sqlx::query(
-            r#"
-            SELECT remote_endpoint_id, generation, issued_grant_id, held_grant_id, revoked_at
-            FROM relationship_generation_tombstones
-            WHERE remote_endpoint_id = ?1 AND generation = ?2
-            "#,
-        )
-        .bind(peer_endpoint_id)
-        .bind(generation as i64)
-        .fetch_optional(&self.pool)
-        .await
-        .map_err(VnidropError::repository)?;
-        Ok(row.map(|row| GenerationTombstone {
-            remote_endpoint_id: row.get("remote_endpoint_id"),
-            generation: row.get::<i64, _>("generation") as u64,
-            issued_grant_id: row.get("issued_grant_id"),
-            held_grant_id: row.get("held_grant_id"),
-            revoked_at: row.get("revoked_at"),
-        }))
+    ) -> Result<Option<super::store::GenerationTombstone>, VnidropError> {
+        self.store
+            .find_tombstone(peer_endpoint_id, generation)
+            .await
     }
 
     async fn clear_grant_secrets(&self, row: &RelationshipRow) -> Result<(), VnidropError> {

crates/vnidrop/src/device_relationship/protocol.rs

@@ -0,0 +1,226 @@
+//! Iroh ALPN handler and client for mutual-consent pairing.
+//!
+//! Wire messages and transport live here; durable state and grant custody stay on
+//! [`super::service::DeviceRelationshipService`].
+
+use std::{fmt, sync::Arc};
+
+use iroh::{
+    endpoint::Connection,
+    protocol::{AcceptError, ProtocolHandler},
+    Endpoint, EndpointAddr,
+};
+use irpc::{channel::oneshot, rpc_requests, Client, WithChannels};
+use irpc_iroh::{read_request, IrohLazyRemoteConnection};
+use serde::{Deserialize, Serialize};
+
+use super::DeviceRelationshipService;
+
+#[derive(Clone)]
+pub(crate) struct RelationshipProtocol {
+    relationships: Arc<DeviceRelationshipService>,
+}
+
+impl RelationshipProtocol {
+    pub(crate) const ALPN: &'static [u8] = b"/vnidrop/relationship/1";
+
+    pub(crate) fn new(relationships: Arc<DeviceRelationshipService>) -> Self {
+        Self { relationships }
+    }
+}
+
+impl fmt::Debug for RelationshipProtocol {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str("RelationshipProtocol")
+    }
+}
+
+impl ProtocolHandler for RelationshipProtocol {
+    async fn accept(&self, connection: Connection) -> Result<(), AcceptError> {
+        let remote_endpoint_id = connection.remote_id().to_string();
+        while let Some(message) = read_request::<RelationshipMessages>(&connection).await? {
+            match message {
+                RelationshipMessage::PairingRequest(message) => {
+                    let WithChannels { inner, tx, .. } = message;
+                    let response = self
+                        .relationships
+                        .handle_pairing_request(remote_endpoint_id.clone(), inner)
+                        .await;
+                    let _ = tx.send(response).await;
+                }
+                RelationshipMessage::PairingConsent(message) => {
+                    let WithChannels { inner, tx, .. } = message;
+                    let response = self
+                        .relationships
+                        .handle_pairing_consent(remote_endpoint_id.clone(), inner)
+                        .await;
+                    let _ = tx.send(response).await;
+                }
+                RelationshipMessage::PairingAck(message) => {
+                    let WithChannels { inner, tx, .. } = message;
+                    let response = self
+                        .relationships
+                        .handle_pairing_ack(remote_endpoint_id.clone(), inner)
+                        .await;
+                    let _ = tx.send(response).await;
+                }
+                RelationshipMessage::RevokeNotice(message) => {
+                    let WithChannels { inner, tx, .. } = message;
+                    let acknowledged = self
+                        .relationships
+                        .handle_remote_revoke(remote_endpoint_id.clone(), inner.generation)
+                        .await;
+                    let response = if acknowledged {
+                        RevokeNoticeResponse::Acknowledged
+                    } else {
+                        RevokeNoticeResponse::Rejected
+                    };
+                    let _ = tx.send(response).await;
+                }
+            }
+        }
+        connection.closed().await;
+        Ok(())
+    }
+}
+
+pub(super) struct RelationshipClient {
+    inner: Client<RelationshipMessages>,
+}
+
+impl RelationshipClient {
+    pub(super) fn connect(endpoint: Endpoint, addr: EndpointAddr) -> Self {
+        Self {
+            inner: Client::boxed(IrohLazyRemoteConnection::new(
+                endpoint,
+                addr,
+                RelationshipProtocol::ALPN.to_vec(),
+            )),
+        }
+    }
+
+    pub(super) async fn pairing_request(
+        &self,
+        request: PairingRequest,
+    ) -> Result<PairingRequestResponse, irpc::Error> {
+        self.inner.rpc(request).await
+    }
+
+    pub(super) async fn pairing_consent(
+        &self,
+        consent: PairingConsent,
+    ) -> Result<PairingConsentResponse, irpc::Error> {
+        self.inner.rpc(consent).await
+    }
+
+    pub(super) async fn pairing_ack(
+        &self,
+        ack: PairingAck,
+    ) -> Result<PairingAckResponse, irpc::Error> {
+        self.inner.rpc(ack).await
+    }
+
+    pub(super) async fn revoke_notice(
+        &self,
+        notice: RevokeNotice,
+    ) -> Result<RevokeNoticeResponse, irpc::Error> {
+        self.inner.rpc(notice).await
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct PairingRequest {
+    pub(super) session_id: String,
+    pub(super) capability: Vec<u8>,
+    pub(super) protocol_version: u16,
+    pub(super) generation: u64,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub(crate) enum PairingRequestResponse {
+    AwaitingConsent,
+    Merged,
+    AlreadySaved,
+    Rejected,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct PairingConsent {
+    pub(super) accepted: bool,
+    pub(super) grant: Option<WireGrant>,
+    pub(super) challenge: Option<String>,
+    pub(super) generation: u64,
+    pub(super) protocol_version: u16,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) enum PairingConsentResponse {
+    Completed {
+        grant: Box<WireGrant>,
+        possession_proof: WireProof,
+        ack_challenge: String,
+    },
+    AlreadySaved,
+    Rejected,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct PairingAck {
+    pub(super) possession_proof: WireProof,
+    pub(super) challenge: String,
+    pub(super) generation: u64,
+    pub(super) protocol_version: u16,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub(crate) enum PairingAckResponse {
+    Acknowledged,
+    AlreadySaved,
+    Rejected,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct WireGrant {
+    pub(super) grant_id: String,
+    pub(super) secret: String,
+    pub(super) issuer_endpoint_id: String,
+    pub(super) holder_endpoint_id: String,
+    pub(super) generation: u64,
+    pub(super) protocol_version: u16,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct WireProof {
+    pub(crate) grant_id: String,
+    pub(crate) mac: String,
+    pub(crate) challenge: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct RevokeNotice {
+    pub(super) generation: u64,
+    pub(super) issued_grant_id: Option<String>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub(crate) enum RevokeNoticeResponse {
+    Acknowledged,
+    Rejected,
+}
+
+#[rpc_requests(message = RelationshipMessage)]
+#[derive(Debug, Serialize, Deserialize)]
+#[allow(
+    clippy::enum_variant_names,
+    reason = "Pairing* names mirror the wire RPC surface"
+)]
+enum RelationshipMessages {
+    #[rpc(tx = oneshot::Sender<PairingRequestResponse>)]
+    PairingRequest(PairingRequest),
+    #[rpc(tx = oneshot::Sender<PairingConsentResponse>)]
+    PairingConsent(PairingConsent),
+    #[rpc(tx = oneshot::Sender<PairingAckResponse>)]
+    PairingAck(PairingAck),
+    #[rpc(tx = oneshot::Sender<RevokeNoticeResponse>)]
+    RevokeNotice(RevokeNotice),
+}

crates/vnidrop/src/device_relationship/store.rs

@@ -0,0 +1,574 @@
+//! Durable device-relationship rows (schema + queries).
+//!
+//! Orchestration (custody, pairing RPC, events) stays on
+//! [`super::DeviceRelationshipService`]; this store is the domain adapter held
+//! in [`crate::persistence::AppDataStores`].
+
+use sqlx::{Row, SqlitePool};
+
+use crate::{
+    api::{DeviceRelationship, DeviceRelationshipState, SavedDevice},
+    error::VnidropError,
+    util::now_ms,
+};
+
+/// Minimal non-secret tombstone for a revoked relationship generation.
+///
+/// Retains only what is needed to reject replay: peer identity, generation,
+/// opaque grant ids, and revocation time. No names, filenames, history, or
+/// capability material.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct GenerationTombstone {
+    pub(crate) remote_endpoint_id: String,
+    pub(crate) generation: u64,
+    pub(crate) issued_grant_id: Option<String>,
+    pub(crate) held_grant_id: Option<String>,
+    pub(crate) revoked_at: i64,
+}
+
+#[derive(Debug, Clone)]
+pub(super) struct RelationshipRow {
+    pub(super) state: DeviceRelationshipState,
+    pub(super) generation: u64,
+    pub(super) minimum_protocol_version: u16,
+    pub(super) session_id: Option<String>,
+    pub(super) issued_grant_handle: Option<String>,
+    pub(super) held_grant_handle: Option<String>,
+    pub(super) issued_grant_id: Option<String>,
+    pub(super) held_grant_id: Option<String>,
+    pub(super) created_at: i64,
+}
+
+/// Compact projection used by grant-secret reconcile.
+#[derive(Debug, Clone)]
+pub(super) struct ReconcileRow {
+    pub(super) remote_endpoint_id: String,
+    pub(super) state: DeviceRelationshipState,
+    pub(super) issued_grant_handle: Option<String>,
+    pub(super) held_grant_handle: Option<String>,
+}
+
+pub(super) struct RelationshipUpsert<'a> {
+    pub(super) remote_endpoint_id: &'a str,
+    pub(super) state: DeviceRelationshipState,
+    pub(super) generation: u64,
+    pub(super) minimum_protocol_version: u16,
+    pub(super) session_id: Option<&'a str>,
+    pub(super) issued_grant_handle: Option<&'a str>,
+    pub(super) held_grant_handle: Option<&'a str>,
+    pub(super) issued_grant_id: Option<&'a str>,
+    pub(super) held_grant_id: Option<&'a str>,
+    pub(super) peer_ack: bool,
+    pub(super) local_ack: bool,
+    pub(super) created_at: i64,
+    pub(super) updated_at: i64,
+}
+
+/// Domain store for `device_relationships` (+ generation tombstones).
+#[derive(Clone)]
+pub(crate) struct DeviceRelationshipStore {
+    pool: SqlitePool,
+}
+
+impl DeviceRelationshipStore {
+    pub(crate) fn new(pool: SqlitePool) -> Self {
+        Self { pool }
+    }
+
+    pub(crate) async fn ensure_schema(pool: &SqlitePool) -> anyhow::Result<()> {
+        sqlx::query(
+            r#"
+            CREATE TABLE IF NOT EXISTS device_relationships (
+                remote_endpoint_id TEXT PRIMARY KEY,
+                state TEXT NOT NULL,
+                generation INTEGER NOT NULL,
+                minimum_protocol_version INTEGER NOT NULL,
+                session_id TEXT,
+                issued_grant_handle TEXT,
+                held_grant_handle TEXT,
+                issued_grant_id TEXT,
+                held_grant_id TEXT,
+                peer_ack INTEGER NOT NULL DEFAULT 0,
+                local_ack INTEGER NOT NULL DEFAULT 0,
+                created_at INTEGER NOT NULL,
+                updated_at INTEGER NOT NULL
+            );
+            "#,
+        )
+        .execute(pool)
+        .await?;
+        let columns = sqlx::query("PRAGMA table_info(device_relationships)")
+            .fetch_all(pool)
+            .await?;
+        let has = |name: &str| columns.iter().any(|row| row.get::<String, _>(1) == name);
+        if !has("issued_grant_id") {
+            sqlx::query("ALTER TABLE device_relationships ADD COLUMN issued_grant_id TEXT")
+                .execute(pool)
+                .await?;
+        }
+        if !has("held_grant_id") {
+            sqlx::query("ALTER TABLE device_relationships ADD COLUMN held_grant_id TEXT")
+                .execute(pool)
+                .await?;
+        }
+        if !has("local_label") {
+            sqlx::query("ALTER TABLE device_relationships ADD COLUMN local_label TEXT")
+                .execute(pool)
+                .await?;
+        }
+        sqlx::query(
+            r#"
+            CREATE TABLE IF NOT EXISTS relationship_generation_tombstones (
+                remote_endpoint_id TEXT NOT NULL,
+                generation INTEGER NOT NULL,
+                issued_grant_id TEXT,
+                held_grant_id TEXT,
+                revoked_at INTEGER NOT NULL,
+                PRIMARY KEY (remote_endpoint_id, generation)
+            );
+            "#,
+        )
+        .execute(pool)
+        .await?;
+        Ok(())
+    }
+
+    pub(super) async fn count_active_slots(&self) -> Result<u64, VnidropError> {
+        let row = sqlx::query(
+            r#"
+            SELECT COUNT(*) AS n FROM device_relationships
+            WHERE state IN ('saved', 'pending_outgoing', 'pending_incoming')
+            "#,
+        )
+        .fetch_one(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        Ok(row.get::<i64, _>("n") as u64)
+    }
+
+    pub(super) async fn list_reconcile_rows(&self) -> Result<Vec<ReconcileRow>, VnidropError> {
+        let rows = sqlx::query(
+            r#"
+            SELECT remote_endpoint_id, issued_grant_handle, held_grant_handle, state
+            FROM device_relationships
+            "#,
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        rows.into_iter()
+            .map(|row| {
+                Ok(ReconcileRow {
+                    remote_endpoint_id: row.get("remote_endpoint_id"),
+                    state: parse_state(&row.get::<String, _>("state"))?,
+                    issued_grant_handle: row.get("issued_grant_handle"),
+                    held_grant_handle: row.get("held_grant_handle"),
+                })
+            })
+            .collect()
+    }
+
+    pub(super) async fn list(&self) -> Result<Vec<DeviceRelationship>, VnidropError> {
+        let rows = sqlx::query(
+            r#"
+            SELECT remote_endpoint_id, state, generation, minimum_protocol_version, created_at, updated_at
+            FROM device_relationships
+            ORDER BY updated_at DESC
+            "#,
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        rows.into_iter().map(row_to_relationship).collect()
+    }
+
+    pub(super) async fn list_saved_devices(&self) -> Result<Vec<SavedDevice>, VnidropError> {
+        let rows = sqlx::query(
+            r#"
+            SELECT remote_endpoint_id, local_label, created_at, updated_at
+            FROM device_relationships
+            WHERE state = 'saved'
+            ORDER BY updated_at DESC
+            "#,
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        Ok(rows
+            .into_iter()
+            .map(|row| SavedDevice {
+                endpoint_id: row.get("remote_endpoint_id"),
+                local_label: row.get("local_label"),
+                remote_display_name: None,
+                created_at: row.get("created_at"),
+                last_authenticated_at: Some(row.get("updated_at")),
+            })
+            .collect())
+    }
+
+    pub(super) async fn set_saved_device_label(
+        &self,
+        peer_endpoint_id: &str,
+        label: Option<String>,
+    ) -> Result<bool, VnidropError> {
+        let result = sqlx::query(
+            r#"
+            UPDATE device_relationships
+            SET local_label = ?2, updated_at = ?3
+            WHERE remote_endpoint_id = ?1 AND state = 'saved'
+            "#,
+        )
+        .bind(peer_endpoint_id)
+        .bind(label)
+        .bind(now_ms())
+        .execute(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        Ok(result.rows_affected() > 0)
+    }
+
+    pub(super) async fn set_issued_grant(
+        &self,
+        peer_endpoint_id: &str,
+        handle: &str,
+        grant_id: &str,
+    ) -> Result<(), VnidropError> {
+        sqlx::query(
+            r#"
+            UPDATE device_relationships
+            SET issued_grant_handle = ?2, issued_grant_id = ?3, updated_at = ?4
+            WHERE remote_endpoint_id = ?1
+            "#,
+        )
+        .bind(peer_endpoint_id)
+        .bind(handle)
+        .bind(grant_id)
+        .bind(now_ms())
+        .execute(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        Ok(())
+    }
+
+    pub(super) async fn set_held_grant(
+        &self,
+        peer_endpoint_id: &str,
+        handle: &str,
+        grant_id: &str,
+    ) -> Result<(), VnidropError> {
+        sqlx::query(
+            r#"
+            UPDATE device_relationships
+            SET held_grant_handle = ?2, held_grant_id = ?3, updated_at = ?4
+            WHERE remote_endpoint_id = ?1
+            "#,
+        )
+        .bind(peer_endpoint_id)
+        .bind(handle)
+        .bind(grant_id)
+        .bind(now_ms())
+        .execute(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        Ok(())
+    }
+
+    #[cfg(test)]
+    pub(super) async fn set_minimum_protocol_version(
+        &self,
+        peer_endpoint_id: &str,
+        minimum_protocol_version: u16,
+    ) -> Result<(), VnidropError> {
+        sqlx::query(
+            "UPDATE device_relationships SET minimum_protocol_version = ?2, updated_at = ?3 WHERE remote_endpoint_id = ?1",
+        )
+        .bind(peer_endpoint_id)
+        .bind(i64::from(minimum_protocol_version))
+        .bind(now_ms())
+        .execute(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        Ok(())
+    }
+
+    pub(super) async fn set_acks(
+        &self,
+        peer_endpoint_id: &str,
+        local_ack: bool,
+        peer_ack: bool,
+    ) -> Result<(), VnidropError> {
+        sqlx::query(
+            "UPDATE device_relationships SET local_ack = ?2, peer_ack = ?3, updated_at = ?4 WHERE remote_endpoint_id = ?1",
+        )
+        .bind(peer_endpoint_id)
+        .bind(i64::from(local_ack))
+        .bind(i64::from(peer_ack))
+        .bind(now_ms())
+        .execute(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        Ok(())
+    }
+
+    pub(super) async fn set_state(
+        &self,
+        peer_endpoint_id: &str,
+        state: DeviceRelationshipState,
+    ) -> Result<(), VnidropError> {
+        sqlx::query(
+            "UPDATE device_relationships SET state = ?2, updated_at = ?3 WHERE remote_endpoint_id = ?1",
+        )
+        .bind(peer_endpoint_id)
+        .bind(state_as_str(state))
+        .bind(now_ms())
+        .execute(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        Ok(())
+    }
+
+    pub(super) async fn list_expired_pending_peers(
+        &self,
+        cutoff_ms: i64,
+    ) -> Result<Vec<String>, VnidropError> {
+        let rows = sqlx::query(
+            r#"
+            SELECT remote_endpoint_id FROM device_relationships
+            WHERE state IN ('pending_outgoing', 'pending_incoming') AND updated_at < ?1
+            "#,
+        )
+        .bind(cutoff_ms)
+        .fetch_all(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        Ok(rows.into_iter().map(|row| row.get(0)).collect())
+    }
+
+    pub(super) async fn delete(&self, peer_endpoint_id: &str) -> Result<(), VnidropError> {
+        sqlx::query("DELETE FROM device_relationships WHERE remote_endpoint_id = ?1")
+            .bind(peer_endpoint_id)
+            .execute(&self.pool)
+            .await
+            .map_err(VnidropError::repository)?;
+        Ok(())
+    }
+
+    pub(super) async fn find_row(
+        &self,
+        peer_endpoint_id: &str,
+    ) -> Result<Option<RelationshipRow>, VnidropError> {
+        let row = sqlx::query(
+            r#"
+            SELECT remote_endpoint_id, state, generation, minimum_protocol_version, session_id,
+                   issued_grant_handle, held_grant_handle, issued_grant_id, held_grant_id,
+                   peer_ack, local_ack, created_at, updated_at
+            FROM device_relationships WHERE remote_endpoint_id = ?1
+            "#,
+        )
+        .bind(peer_endpoint_id)
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        row.map(relationship_row_from_sql).transpose()
+    }
+
+    pub(super) async fn upsert(&self, entry: RelationshipUpsert<'_>) -> Result<(), VnidropError> {
+        sqlx::query(
+            r#"
+            INSERT INTO device_relationships (
+                remote_endpoint_id, state, generation, minimum_protocol_version, session_id,
+                issued_grant_handle, held_grant_handle, issued_grant_id, held_grant_id,
+                peer_ack, local_ack, created_at, updated_at
+            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)
+            ON CONFLICT(remote_endpoint_id) DO UPDATE SET
+                state = excluded.state,
+                generation = excluded.generation,
+                minimum_protocol_version = excluded.minimum_protocol_version,
+                session_id = excluded.session_id,
+                issued_grant_handle = COALESCE(excluded.issued_grant_handle, device_relationships.issued_grant_handle),
+                held_grant_handle = COALESCE(excluded.held_grant_handle, device_relationships.held_grant_handle),
+                issued_grant_id = COALESCE(excluded.issued_grant_id, device_relationships.issued_grant_id),
+                held_grant_id = COALESCE(excluded.held_grant_id, device_relationships.held_grant_id),
+                peer_ack = excluded.peer_ack,
+                local_ack = excluded.local_ack,
+                updated_at = excluded.updated_at
+            "#,
+        )
+        .bind(entry.remote_endpoint_id)
+        .bind(state_as_str(entry.state))
+        .bind(entry.generation as i64)
+        .bind(i64::from(entry.minimum_protocol_version))
+        .bind(entry.session_id)
+        .bind(entry.issued_grant_handle)
+        .bind(entry.held_grant_handle)
+        .bind(entry.issued_grant_id)
+        .bind(entry.held_grant_id)
+        .bind(i64::from(entry.peer_ack))
+        .bind(i64::from(entry.local_ack))
+        .bind(entry.created_at)
+        .bind(entry.updated_at)
+        .execute(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        Ok(())
+    }
+
+    /// Bump generation and clear grant columns after a prior generation was tombstoned.
+    pub(super) async fn begin_grant_rotation(
+        &self,
+        peer_endpoint_id: &str,
+        new_generation: u64,
+    ) -> Result<(), VnidropError> {
+        sqlx::query(
+            r#"
+            UPDATE device_relationships
+            SET generation = ?2,
+                issued_grant_handle = NULL,
+                held_grant_handle = NULL,
+                issued_grant_id = NULL,
+                held_grant_id = NULL,
+                updated_at = ?3
+            WHERE remote_endpoint_id = ?1
+            "#,
+        )
+        .bind(peer_endpoint_id)
+        .bind(new_generation as i64)
+        .bind(now_ms())
+        .execute(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        Ok(())
+    }
+
+    pub(super) async fn insert_tombstone(
+        &self,
+        peer_endpoint_id: &str,
+        row: &RelationshipRow,
+    ) -> Result<(), VnidropError> {
+        sqlx::query(
+            r#"
+            INSERT INTO relationship_generation_tombstones (
+                remote_endpoint_id, generation, issued_grant_id, held_grant_id, revoked_at
+            ) VALUES (?1, ?2, ?3, ?4, ?5)
+            ON CONFLICT(remote_endpoint_id, generation) DO UPDATE SET
+                issued_grant_id = COALESCE(excluded.issued_grant_id, relationship_generation_tombstones.issued_grant_id),
+                held_grant_id = COALESCE(excluded.held_grant_id, relationship_generation_tombstones.held_grant_id),
+                revoked_at = excluded.revoked_at
+            "#,
+        )
+        .bind(peer_endpoint_id)
+        .bind(row.generation as i64)
+        .bind(row.issued_grant_id.as_deref())
+        .bind(row.held_grant_id.as_deref())
+        .bind(now_ms())
+        .execute(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        Ok(())
+    }
+
+    pub(super) async fn find_tombstone(
+        &self,
+        peer_endpoint_id: &str,
+        generation: u64,
+    ) -> Result<Option<GenerationTombstone>, VnidropError> {
+        let row = sqlx::query(
+            r#"
+            SELECT remote_endpoint_id, generation, issued_grant_id, held_grant_id, revoked_at
+            FROM relationship_generation_tombstones
+            WHERE remote_endpoint_id = ?1 AND generation = ?2
+            "#,
+        )
+        .bind(peer_endpoint_id)
+        .bind(generation as i64)
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        Ok(row.map(|row| GenerationTombstone {
+            remote_endpoint_id: row.get("remote_endpoint_id"),
+            generation: row.get::<i64, _>("generation") as u64,
+            issued_grant_id: row.get("issued_grant_id"),
+            held_grant_id: row.get("held_grant_id"),
+            revoked_at: row.get("revoked_at"),
+        }))
+    }
+
+    #[cfg(test)]
+    pub(crate) async fn list_tombstones(
+        &self,
+        peer_endpoint_id: &str,
+    ) -> Result<Vec<GenerationTombstone>, VnidropError> {
+        let rows = sqlx::query(
+            r#"
+            SELECT remote_endpoint_id, generation, issued_grant_id, held_grant_id, revoked_at
+            FROM relationship_generation_tombstones
+            WHERE remote_endpoint_id = ?1
+            ORDER BY generation ASC
+            "#,
+        )
+        .bind(peer_endpoint_id)
+        .fetch_all(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        Ok(rows
+            .into_iter()
+            .map(|row| GenerationTombstone {
+                remote_endpoint_id: row.get("remote_endpoint_id"),
+                generation: row.get::<i64, _>("generation") as u64,
+                issued_grant_id: row.get("issued_grant_id"),
+                held_grant_id: row.get("held_grant_id"),
+                revoked_at: row.get("revoked_at"),
+            })
+            .collect())
+    }
+}
+
+pub(super) fn state_as_str(state: DeviceRelationshipState) -> &'static str {
+    match state {
+        DeviceRelationshipState::PendingOutgoing => "pending_outgoing",
+        DeviceRelationshipState::PendingIncoming => "pending_incoming",
+        DeviceRelationshipState::Saved => "saved",
+        DeviceRelationshipState::Revoked => "revoked",
+        DeviceRelationshipState::Blocked => "blocked",
+    }
+}
+
+fn parse_state(value: &str) -> Result<DeviceRelationshipState, VnidropError> {
+    match value {
+        "pending_outgoing" => Ok(DeviceRelationshipState::PendingOutgoing),
+        "pending_incoming" => Ok(DeviceRelationshipState::PendingIncoming),
+        "saved" => Ok(DeviceRelationshipState::Saved),
+        "revoked" => Ok(DeviceRelationshipState::Revoked),
+        "blocked" => Ok(DeviceRelationshipState::Blocked),
+        _ => Err(VnidropError::Internal {
+            reason: "unknown device relationship state".to_string(),
+        }),
+    }
+}
+
+fn row_to_relationship(row: sqlx::sqlite::SqliteRow) -> Result<DeviceRelationship, VnidropError> {
+    Ok(DeviceRelationship {
+        remote_endpoint_id: row.get("remote_endpoint_id"),
+        state: parse_state(&row.get::<String, _>("state"))?,
+        generation: row.get::<i64, _>("generation") as u64,
+        minimum_protocol_version: row.get::<i64, _>("minimum_protocol_version") as u16,
+        created_at: row.get("created_at"),
+        updated_at: row.get("updated_at"),
+    })
+}
+
+fn relationship_row_from_sql(
+    row: sqlx::sqlite::SqliteRow,
+) -> Result<RelationshipRow, VnidropError> {
+    Ok(RelationshipRow {
+        state: parse_state(&row.get::<String, _>("state"))?,
+        generation: row.get::<i64, _>("generation") as u64,
+        minimum_protocol_version: row.get::<i64, _>("minimum_protocol_version") as u16,
+        session_id: row.get("session_id"),
+        issued_grant_handle: row.get("issued_grant_handle"),
+        held_grant_handle: row.get("held_grant_handle"),
+        issued_grant_id: row.get("issued_grant_id"),
+        held_grant_id: row.get("held_grant_id"),
+        created_at: row.get("created_at"),
+    })
+}