feat(core): add event revisions and saved-device local labels

Platform contract harnesses need monotonic event revisions for at-least-once
dedup and a typed rename API that survives listing.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
2026-08-11 04:47:28 +02:00
parent 89c2206f0f
commit 8e8cab9b24
13 changed files with 162 additions and 19 deletions

View File

@@ -70,7 +70,7 @@ final class ProgressDerivationTests: XCTestCase {
private func event(phase: String, kind: String, json: String) -> CoreEventModel { private func event(phase: String, kind: String, json: String) -> CoreEventModel {
CoreEventModel( CoreEventModel(
id: UUID().uuidString, timestamp: 0, scope: "transfer", transferId: 1, id: UUID().uuidString, revision: 1, timestamp: 0, scope: "transfer", transferId: 1,
direction: "send", phase: phase, kind: kind, dataJson: json direction: "send", phase: phase, kind: kind, dataJson: json
) )
} }

View File

@@ -12,6 +12,7 @@ struct CoreStatus: Equatable, Sendable {
struct CoreEventModel: Equatable, Identifiable, Sendable { struct CoreEventModel: Equatable, Identifiable, Sendable {
let id: String let id: String
let revision: UInt64
let timestamp: Int64 let timestamp: Int64
let scope: String let scope: String
let transferId: UInt64? let transferId: UInt64?

View File

@@ -521,7 +521,7 @@ private func withSecurityScopedAccess<T>(pathOrUrl: String, _ body: () throws ->
private extension CoreEvent { private extension CoreEvent {
func toModel() -> CoreEventModel { func toModel() -> CoreEventModel {
CoreEventModel( CoreEventModel(
id: id, timestamp: timestamp, scope: scope, transferId: transferId, id: id, revision: revision, timestamp: timestamp, scope: scope, transferId: transferId,
direction: direction, phase: phase, kind: kind, dataJson: dataJson direction: direction, phase: phase, kind: kind, dataJson: dataJson
) )
} }

View File

@@ -404,6 +404,8 @@ pub fn default_core_limits() -> CoreLimits {
#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)] #[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)]
pub struct CoreEvent { pub struct CoreEvent {
pub id: String, pub id: String,
/// Monotonic per-process revision for at-least-once delivery deduplication.
pub revision: u64,
pub timestamp: i64, pub timestamp: i64,
pub scope: String, pub scope: String,
pub transfer_id: Option<u64>, pub transfer_id: Option<u64>,

View File

@@ -162,6 +162,11 @@ impl DeviceRelationshipService {
.execute(pool) .execute(pool)
.await?; .await?;
} }
if !has("local_label") {
sqlx::query("ALTER TABLE device_relationships ADD COLUMN local_label TEXT")
.execute(pool)
.await?;
}
Self::ensure_lifecycle_schema(pool).await?; Self::ensure_lifecycle_schema(pool).await?;
Ok(()) Ok(())
} }
@@ -252,21 +257,58 @@ impl DeviceRelationshipService {
} }
pub(crate) async fn list_saved_devices(&self) -> Result<Vec<SavedDevice>, VnidropError> { pub(crate) async fn list_saved_devices(&self) -> Result<Vec<SavedDevice>, VnidropError> {
Ok(self let rows = sqlx::query(
.list() r#"
.await? 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() .into_iter()
.filter(|entry| entry.state == DeviceRelationshipState::Saved) .map(|row| SavedDevice {
.map(|entry| SavedDevice { endpoint_id: row.get("remote_endpoint_id"),
endpoint_id: entry.remote_endpoint_id, local_label: row.get("local_label"),
local_label: None,
remote_display_name: None, remote_display_name: None,
created_at: entry.created_at, created_at: row.get("created_at"),
last_authenticated_at: Some(entry.updated_at), last_authenticated_at: Some(row.get("updated_at")),
}) })
.collect()) .collect())
} }
/// Sets the user-owned local label for a Saved device. Labels are never
/// overwritten by remote display names.
pub(crate) async fn set_saved_device_label(
&self,
peer_endpoint_id: String,
label: Option<String>,
) -> Result<(), 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)?;
if result.rows_affected() == 0 {
return Err(VnidropError::invalid_input(anyhow::anyhow!(
"peer is not a saved device"
)));
}
self.emit_changed(&peer_endpoint_id, DeviceRelationshipState::Saved);
Ok(())
}
pub(crate) async fn request_pairing( pub(crate) async fn request_pairing(
self: &Arc<Self>, self: &Arc<Self>,
peer_endpoint_id: String, peer_endpoint_id: String,

View File

@@ -247,8 +247,9 @@ impl EventHub {
.sequence .sequence
.lock() .lock()
.unwrap_or_else(std::sync::PoisonError::into_inner); .unwrap_or_else(std::sync::PoisonError::into_inner);
let id = format!("{timestamp}-{}", *sequence); let revision = *sequence;
*sequence += 1; let id = format!("{timestamp}-{revision}");
*sequence = sequence.saturating_add(1);
drop(sequence); drop(sequence);
// Compose observes this event synchronously, while SQLite persistence is // Compose observes this event synchronously, while SQLite persistence is
@@ -258,6 +259,7 @@ impl EventHub {
// typed UniFFI list APIs still expose the values the UI needs. // typed UniFFI list APIs still expose the values the UI needs.
let event = CoreEvent { let event = CoreEvent {
id, id,
revision,
timestamp, timestamp,
scope: scope.as_str().to_string(), scope: scope.as_str().to_string(),
transfer_id, transfer_id,

View File

@@ -227,6 +227,7 @@ impl Repository {
r#" r#"
CREATE TABLE IF NOT EXISTS transfer_events ( CREATE TABLE IF NOT EXISTS transfer_events (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
revision INTEGER NOT NULL DEFAULT 0,
timestamp INTEGER NOT NULL, timestamp INTEGER NOT NULL,
scope TEXT NOT NULL, scope TEXT NOT NULL,
transfer_id INTEGER, transfer_id INTEGER,
@@ -240,6 +241,20 @@ impl Repository {
.execute(&self.pool) .execute(&self.pool)
.await?; .await?;
let event_columns = sqlx::query("PRAGMA table_info(transfer_events)")
.fetch_all(&self.pool)
.await?;
if !event_columns
.iter()
.any(|row| row.get::<String, _>(1) == "revision")
{
sqlx::query(
"ALTER TABLE transfer_events ADD COLUMN revision INTEGER NOT NULL DEFAULT 0",
)
.execute(&self.pool)
.await?;
}
sqlx::query( sqlx::query(
"CREATE INDEX IF NOT EXISTS idx_transfer_events_transfer_id ON transfer_events(transfer_id, timestamp);", "CREATE INDEX IF NOT EXISTS idx_transfer_events_transfer_id ON transfer_events(transfer_id, timestamp);",
) )
@@ -981,12 +996,13 @@ impl Repository {
sqlx::query( sqlx::query(
r#" r#"
INSERT OR REPLACE INTO transfer_events ( INSERT OR REPLACE INTO transfer_events (
id, timestamp, scope, transfer_id, direction, phase, kind, data_json id, revision, timestamp, scope, transfer_id, direction, phase, kind, data_json
) )
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8); VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9);
"#, "#,
) )
.bind(&event.id) .bind(&event.id)
.bind(to_db_id(event.revision)?)
.bind(event.timestamp) .bind(event.timestamp)
.bind(&event.scope) .bind(&event.scope)
.bind(event.transfer_id.map(to_db_id).transpose()?) .bind(event.transfer_id.map(to_db_id).transpose()?)
@@ -1330,10 +1346,10 @@ impl Repository {
let rows = if let Some(transfer_id) = transfer_id { let rows = if let Some(transfer_id) = transfer_id {
sqlx::query( sqlx::query(
r#" r#"
SELECT id, timestamp, scope, transfer_id, direction, phase, kind, data_json SELECT id, revision, timestamp, scope, transfer_id, direction, phase, kind, data_json
FROM transfer_events FROM transfer_events
WHERE transfer_id = ?1 WHERE transfer_id = ?1
ORDER BY timestamp ASC ORDER BY timestamp ASC, revision ASC
LIMIT ?2 LIMIT ?2
"#, "#,
) )
@@ -1344,9 +1360,9 @@ impl Repository {
} else { } else {
sqlx::query( sqlx::query(
r#" r#"
SELECT id, timestamp, scope, transfer_id, direction, phase, kind, data_json SELECT id, revision, timestamp, scope, transfer_id, direction, phase, kind, data_json
FROM transfer_events FROM transfer_events
ORDER BY timestamp DESC ORDER BY timestamp DESC, revision DESC
LIMIT ?1 LIMIT ?1
"#, "#,
) )
@@ -1413,6 +1429,7 @@ fn row_to_transfer(row: sqlx::sqlite::SqliteRow) -> Result<StoredTransfer> {
fn row_to_event(row: sqlx::sqlite::SqliteRow) -> CoreEvent { fn row_to_event(row: sqlx::sqlite::SqliteRow) -> CoreEvent {
CoreEvent { CoreEvent {
id: row.get("id"), id: row.get("id"),
revision: row.get::<i64, _>("revision") as u64,
timestamp: row.get("timestamp"), timestamp: row.get("timestamp"),
scope: row.get("scope"), scope: row.get("scope"),
transfer_id: row transfer_id: row

View File

@@ -493,6 +493,19 @@ impl VnidropCore {
self.block_on(self.inner.list_saved_devices()) self.block_on(self.inner.list_saved_devices())
} }
/// Sets the user-owned local label for a Saved device.
pub fn set_saved_device_label(
&self,
peer_endpoint_id: String,
label: Option<String>,
) -> Result<(), VnidropError> {
self.block_on(
self.inner
.device_relationships
.set_saved_device_label(peer_endpoint_id, label),
)
}
pub fn respond_to_device_pairing( pub fn respond_to_device_pairing(
&self, &self,
peer_endpoint_id: String, peer_endpoint_id: String,

View File

@@ -540,3 +540,60 @@ fn reinstalled_peer_is_never_merged_by_name_or_metadata() {
assert!(ids.contains(&charlie_id)); assert!(ids.contains(&charlie_id));
assert_eq!(alice.core.list_device_relationships().unwrap().len(), 2); assert_eq!(alice.core.list_device_relationships().unwrap().len(), 2);
} }
#[test]
fn saved_device_local_label_survives_listing_and_rejects_non_saved_peers() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
let bob_id = bob.core.status().endpoint_id.clone();
reach_saved(&alice, &bob, 90_050);
alice
.core
.set_saved_device_label(bob_id.clone(), Some("Kitchen Tablet".to_string()))
.unwrap();
let saved = alice.core.list_saved_devices().unwrap();
assert_eq!(saved.len(), 1);
assert_eq!(saved[0].local_label.as_deref(), Some("Kitchen Tablet"));
alice
.core
.set_saved_device_label(bob_id.clone(), None)
.unwrap();
assert!(alice.core.list_saved_devices().unwrap()[0]
.local_label
.is_none());
let err = alice
.core
.set_saved_device_label("unknown-peer".to_string(), Some("x".to_string()))
.unwrap_err();
assert!(matches!(err, crate::VnidropError::InvalidInput { .. }));
}
#[test]
fn events_carry_stable_ids_and_monotonic_revisions() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
reach_saved(&alice, &bob, 90_051);
let events = alice.core.list_events(None).unwrap();
assert!(!events.is_empty());
let mut seen_ids = std::collections::HashSet::new();
let mut revisions: Vec<u64> = Vec::new();
for event in &events {
assert!(
seen_ids.insert(event.id.clone()),
"event ids must be unique"
);
assert!(event.revision >= 1, "revisions start at 1");
revisions.push(event.revision);
}
revisions.sort_unstable();
revisions.dedup();
assert_eq!(
revisions.len(),
events.len(),
"each event must have a distinct revision"
);
}

View File

@@ -87,6 +87,7 @@ async fn persists_transfers_and_events_across_reopen() {
.insert_event( .insert_event(
&CoreEvent { &CoreEvent {
id: "event-1".to_string(), id: "event-1".to_string(),
revision: 1,
timestamp: 10, timestamp: 10,
scope: "transfer".to_string(), scope: "transfer".to_string(),
transfer_id: Some(7), transfer_id: Some(7),
@@ -663,6 +664,7 @@ async fn event_reads_respect_configured_history_limit() {
.insert_event( .insert_event(
&CoreEvent { &CoreEvent {
id: format!("event-{sequence}"), id: format!("event-{sequence}"),
revision: 1,
timestamp: sequence, timestamp: sequence,
scope: "endpoint".to_string(), scope: "endpoint".to_string(),
transfer_id: None, transfer_id: None,
@@ -711,6 +713,7 @@ async fn deleting_transfer_removes_related_history_transactionally() {
.insert_event( .insert_event(
&CoreEvent { &CoreEvent {
id: "event-delete".to_string(), id: "event-delete".to_string(),
revision: 1,
timestamp: 1, timestamp: 1,
scope: "transfer".to_string(), scope: "transfer".to_string(),
transfer_id: Some(88), transfer_id: Some(88),
@@ -775,6 +778,7 @@ async fn deleting_receive_history_only_removes_terminal_receives_and_dependants(
.insert_event( .insert_event(
&CoreEvent { &CoreEvent {
id: format!("event-{transfer_id}"), id: format!("event-{transfer_id}"),
revision: 1,
timestamp: transfer_id as i64, timestamp: transfer_id as i64,
scope: "transfer".to_string(), scope: "transfer".to_string(),
transfer_id: Some(transfer_id), transfer_id: Some(transfer_id),
@@ -857,6 +861,7 @@ async fn receive_history_mid_transaction_failure_preserves_all_related_rows() {
.insert_event( .insert_event(
&CoreEvent { &CoreEvent {
id: "event-preserved".to_string(), id: "event-preserved".to_string(),
revision: 1,
timestamp: 1, timestamp: 1,
scope: "transfer".to_string(), scope: "transfer".to_string(),
transfer_id: Some(106), transfer_id: Some(106),

View File

@@ -13,6 +13,7 @@ data class CoreStatus(
data class CoreEventModel( data class CoreEventModel(
val id: String, val id: String,
val revision: ULong,
val timestamp: Long, val timestamp: Long,
val scope: String, val scope: String,
val transferId: ULong?, val transferId: ULong?,

View File

@@ -370,6 +370,7 @@ private fun RelaySettings.toNative(): CoreNetworkConfig = when (mode) {
private fun CoreEvent.toModel(): CoreEventModel = CoreEventModel( private fun CoreEvent.toModel(): CoreEventModel = CoreEventModel(
id = id, id = id,
revision = revision,
timestamp = timestamp, timestamp = timestamp,
scope = scope, scope = scope,
transferId = transferId, transferId = transferId,

View File

@@ -180,6 +180,7 @@ class AppUiModelsTest {
), ),
CoreEventModel( CoreEventModel(
id = "conn", id = "conn",
revision = 1UL,
timestamp = 1L, timestamp = 1L,
scope = "endpoint", scope = "endpoint",
transferId = null, transferId = null,
@@ -330,6 +331,7 @@ class AppUiModelsTest {
direction: String = "receive", direction: String = "receive",
) = CoreEventModel( ) = CoreEventModel(
id = id, id = id,
revision = 1UL,
timestamp = 1L, timestamp = 1L,
scope = "transfer", scope = "transfer",
transferId = transferId, transferId = transferId,