diff --git a/crates/buzz-cli/src/commands/tasks.rs b/crates/buzz-cli/src/commands/tasks.rs index 6de86dc1026..ed095070b71 100644 --- a/crates/buzz-cli/src/commands/tasks.rs +++ b/crates/buzz-cli/src/commands/tasks.rs @@ -222,6 +222,7 @@ pub async fn dispatch( clear_assignee, due_at, clear_due, + expected_revision, } => { let task = uuid("task", &task)?; let mut payload = Map::new(); @@ -244,6 +245,9 @@ pub async fn dispatch( } else if let Some(value) = due_at { payload.insert("due_at".into(), Value::String(value)); } + if let Some(value) = expected_revision { + payload.insert("expected_revision".into(), json!(value)); + } if payload.is_empty() { return Err(CliError::Usage( "update requires at least one mutable field".into(), diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index c203d9e746e..93806f22ea2 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -2149,6 +2149,9 @@ pub enum TasksCmd { due_at: Option, #[arg(long, default_value_t = false)] clear_due: bool, + /// HW-017: reject the PATCH with 409 if the task's revision does not match + #[arg(long)] + expected_revision: Option, }, /// Append a progress/comment event; use '-' to read stdin Comment { task: String, body: String }, diff --git a/crates/buzz-db/examples/hw017_migrate.rs b/crates/buzz-db/examples/hw017_migrate.rs new file mode 100644 index 00000000000..51f831a8798 --- /dev/null +++ b/crates/buzz-db/examples/hw017_migrate.rs @@ -0,0 +1,24 @@ +use sqlx::migrate::Migrator; + +#[tokio::main] +async fn main() { + let url = std::env::var("DATABASE_URL").expect("DATABASE_URL"); + let db = sqlx::postgres::PgPoolOptions::new() + .max_connections(2) + .connect(&url) + .await + .expect("connect"); + let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .parent() + .and_then(|p| p.parent()) + .expect("repo root (crates/buzz-db parent x2)"); + let m = Migrator::new(root.join("migrations").as_path()) + .await + .expect("load migrations"); + m.run(&db).await.expect("run migrations"); + let v: i64 = sqlx::query_scalar("SELECT COALESCE(MAX(version),0) FROM _sqlx_migrations") + .fetch_one(&db) + .await + .expect("version"); + println!("MIGRATED_TO={v}"); +} diff --git a/crates/buzz-db/src/error.rs b/crates/buzz-db/src/error.rs index 4f4e6b105c5..e636ebf5771 100644 --- a/crates/buzz-db/src/error.rs +++ b/crates/buzz-db/src/error.rs @@ -45,6 +45,19 @@ pub enum DbError { #[error("invalid data: {0}")] InvalidData(String), + /// A PATCH carried `expected_revision` that does not match the row's current + /// `revision`. The caller must re-fetch and retry. This is the optimistic + /// concurrency guard from HW-017: a stale write must not silently win. + #[error("task {task_id} revision mismatch: expected {expected}, found {actual}")] + StaleRevision { + /// The task that was being patched. + task_id: uuid::Uuid, + /// The revision the caller expected (the snapshot it read). + expected: i32, + /// The revision the row actually carries. + actual: i32, + }, + /// A serving write admitted before the lifecycle transition is still live. /// This is an ordinary retryable drain condition, not a safety violation. #[error( diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 81a16cf0c5e..55ecd760692 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -704,8 +704,10 @@ mod postgres_tests { // upstream carries 44 (0032-0034 and 0040 adopted from our PRs); // fork adds 0046_task_system (PR #6425 pending upstream) and - // 0047_agent_machine_homes (AGENT-HOMES-001 PR-3). - assert_eq!(migrations.len(), 46); + // 0047_agent_machine_homes (AGENT-HOMES-001 PR-3), and + // 0050_task_optimistic_concurrency (HW-017 optimistic concurrency guard). + assert_eq!(migrations.len(), 47); + assert_eq!(migrations[46].version, 50); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] diff --git a/crates/buzz-db/src/task.rs b/crates/buzz-db/src/task.rs index c8b6018b461..1e378e09075 100644 --- a/crates/buzz-db/src/task.rs +++ b/crates/buzz-db/src/task.rs @@ -36,7 +36,7 @@ macro_rules! task_columns { () => { "community_id, id, channel_id, created_by_pubkey, assignee_pubkey, \ parent_task_id, title, body, status, priority, source, source_ref, \ - due_at, done_at, archived_at, created_at, updated_at" + due_at, done_at, archived_at, created_at, updated_at, revision" }; } @@ -82,6 +82,9 @@ pub struct TaskRecord { pub created_at: DateTime, /// Last-modification timestamp. pub updated_at: DateTime, + /// Monotonic revision counter for optimistic concurrency (HW-017). + /// Increments on every UPDATE via trigger; 0 on a fresh row. + pub revision: i32, } /// One entry in a task's append-only history. @@ -168,10 +171,21 @@ pub struct TaskPatch { pub due_at: Option>>, /// New assignee, or `Some(None)` to unassign. pub assignee_pubkey: Option>>, + /// Optimistic concurrency guard (HW-017): the revision the caller + /// believes the task is at. If the row's `revision` does not match, + /// `update_task` returns [`DbError::StaleRevision`] without writing. + /// `None` skips the guard (backward-compatible with pre-HW-017 clients). + pub expected_revision: Option, } impl TaskPatch { /// Whether the patch asks for any change at all. + /// + /// `expected_revision` is deliberately NOT counted: it is a precondition on + /// a change, not a change. Counting it would let a body carrying only + /// `expected_revision` pass the relay's "patch must change at least one + /// field" check and reach `update_task`, which would then restate every + /// column at its current value — a write with no requested change. pub fn is_empty(&self) -> bool { self.status.is_none() && self.title.is_none() @@ -204,6 +218,7 @@ fn parse_task_row(row: &sqlx::postgres::PgRow) -> Result { archived_at: row.try_get("archived_at")?, created_at: row.try_get("created_at")?, updated_at: row.try_get("updated_at")?, + revision: row.try_get("revision")?, }) } @@ -420,6 +435,21 @@ pub async fn update_task( .ok_or_else(|| DbError::NotFound(format!("task {id}")))?; let current = parse_task_row(¤t)?; + // HW-017: optimistic concurrency guard. If the caller supplied an + // expected revision, it must match the row's current revision or the + // PATCH is rejected before any write. A mismatch means another writer + // committed since the caller last fetched the task; letting the stale + // write proceed would silently clobber their change. + if let Some(expected) = patch.expected_revision { + if expected != current.revision { + return Err(DbError::StaleRevision { + task_id: id, + expected, + actual: current.revision, + }); + } + } + let new_status = patch.status.unwrap_or(current.status); let new_title = patch.title.clone().unwrap_or_else(|| current.title.clone()); let new_priority = patch.priority.unwrap_or(current.priority); @@ -438,7 +468,7 @@ pub async fn update_task( let row = sqlx::query(concat!( "UPDATE tasks SET status = $3, title = $4, priority = $5, due_at = $6, \ - assignee_pubkey = $7, done_at = $8, updated_at = NOW() \ + assignee_pubkey = $7, done_at = $8 \ WHERE community_id = $1 AND id = $2 \ RETURNING ", task_columns!() @@ -954,6 +984,346 @@ mod postgres_tests { delete_test_community(&pool, community).await; } + /// HW-017: the revision counter must advance on real change and hold still + /// on a semantic no-op. If a restated value bumped the revision, every + /// other client's `expected_revision` would be invalidated by a write that + /// changed nothing, manufacturing spurious 409s on idempotent retries. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn revision_advances_on_real_change_and_holds_on_a_semantic_noop() { + let pool = setup_pool().await; + let community = make_test_community(&pool).await; + let creator = make_test_user(&pool, community, 0x51).await; + + let task = create_task( + &pool, + community, + NewTask { + created_by_pubkey: Some(creator.clone()), + title: "revision probe".to_owned(), + ..NewTask::default() + }, + ) + .await + .expect("create task"); + assert_eq!(task.revision, 0, "a fresh task starts at revision 0"); + + let bumped = update_task( + &pool, + community, + task.id, + &TaskPatch { + status: Some(TaskStatus::InProgress), + ..TaskPatch::default() + }, + Some(&creator), + ) + .await + .expect("real change"); + assert_eq!(bumped.revision, 1, "a real change bumps exactly once"); + + // Restate the values the row already holds. The trigger fires, but the + // whole-row comparison sees no payload change, so revision must hold. + let restated = update_task( + &pool, + community, + task.id, + &TaskPatch { + status: Some(TaskStatus::InProgress), + title: Some("revision probe".to_owned()), + ..TaskPatch::default() + }, + Some(&creator), + ) + .await + .expect("semantic no-op"); + assert_eq!( + restated.revision, 1, + "a semantic no-op must NOT bump the revision" + ); + assert_eq!( + restated.updated_at, bumped.updated_at, + "a semantic no-op must not touch updated_at either" + ); + + let bumped_again = update_task( + &pool, + community, + task.id, + &TaskPatch { + title: Some("renamed".to_owned()), + ..TaskPatch::default() + }, + Some(&creator), + ) + .await + .expect("second real change"); + assert_eq!( + bumped_again.revision, 2, + "the counter still advances after a no-op" + ); + + delete_test_community(&pool, community).await; + } + + /// HW-017: the guard itself. A patch built from a stale snapshot must be + /// rejected with `StaleRevision` and must leave the row untouched. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn a_stale_expected_revision_is_rejected_and_changes_nothing() { + let pool = setup_pool().await; + let community = make_test_community(&pool).await; + let creator = make_test_user(&pool, community, 0x52).await; + + let task = create_task( + &pool, + community, + NewTask { + created_by_pubkey: Some(creator.clone()), + title: "contended".to_owned(), + ..NewTask::default() + }, + ) + .await + .expect("create task"); + + // Writer A reads revision 0 and commits, moving the row to revision 1. + let winner = update_task( + &pool, + community, + task.id, + &TaskPatch { + title: Some("writer A won".to_owned()), + expected_revision: Some(task.revision), + ..TaskPatch::default() + }, + Some(&creator), + ) + .await + .expect("writer A commits against a fresh snapshot"); + assert_eq!(winner.revision, 1); + + // Writer B still holds the revision-0 snapshot. Its write must lose. + let error = update_task( + &pool, + community, + task.id, + &TaskPatch { + title: Some("writer B clobbers".to_owned()), + expected_revision: Some(task.revision), + ..TaskPatch::default() + }, + Some(&creator), + ) + .await + .expect_err("a stale write must not silently win"); + match error { + DbError::StaleRevision { + task_id, + expected, + actual, + } => { + assert_eq!(task_id, task.id); + assert_eq!(expected, 0, "the snapshot writer B read"); + assert_eq!(actual, 1, "the revision the row actually carries"); + } + other => panic!("expected StaleRevision, got {other:?}"), + } + + // The rejection must be total: writer A's value survives intact. + let after = get_task(&pool, community, task.id).await.expect("re-fetch"); + assert_eq!( + after.title, "writer A won", + "the losing write must not have applied any field" + ); + assert_eq!(after.revision, 1, "a rejected write must not bump"); + + // Re-fetching and retrying against the current revision succeeds. + let retried = update_task( + &pool, + community, + task.id, + &TaskPatch { + title: Some("writer B retried".to_owned()), + expected_revision: Some(after.revision), + ..TaskPatch::default() + }, + Some(&creator), + ) + .await + .expect("retry against the current revision"); + assert_eq!(retried.title, "writer B retried"); + assert_eq!(retried.revision, 2); + + // A patch that omits `expected_revision` keeps the previous + // last-write-wins behaviour, so existing clients are unaffected. + let unguarded = update_task( + &pool, + community, + task.id, + &TaskPatch { + title: Some("unguarded still works".to_owned()), + ..TaskPatch::default() + }, + Some(&creator), + ) + .await + .expect("an unguarded patch is still accepted"); + assert_eq!(unguarded.revision, 3); + + delete_test_community(&pool, community).await; + } + + /// HW-017: the interleaving the guard exists for. Two connections race on + /// one task: writer A holds the `FOR UPDATE` row lock while writer B — on + /// its own pool, blocked mid-`update_task` — waits for that lock. A then + /// commits a real change (revision 0 -> 1). B, whose snapshot predates A's + /// commit, must be rejected with `StaleRevision` once it acquires the lock, + /// not silently overwrite A. Sequential rejection is covered above; this + /// test proves the guard under a genuine overlapping-transaction race. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn a_grounded_writer_loses_against_an_interleaved_commit() { + let a_pool = setup_pool().await; + let b_pool = PgPool::connect(&test_database_url()) + .await + .expect("writer B's own connection"); + let community = make_test_community(&a_pool).await; + let creator = make_test_user(&a_pool, community, 0x61).await; + + let task = create_task( + &a_pool, + community, + NewTask { + created_by_pubkey: Some(creator.clone()), + title: "interleaved".to_owned(), + ..NewTask::default() + }, + ) + .await + .expect("create task"); + + // Writer A opens a transaction and takes the row lock, but does not + // commit yet — it pauses with the lock held. + let mut a_tx = a_pool.begin().await.expect("writer A begin"); + sqlx::query(concat!( + "SELECT ", + task_columns!(), + " FROM tasks WHERE community_id = $1 AND id = $2 FOR UPDATE" + )) + .bind(community.as_uuid()) + .bind(task.id) + .fetch_one(&mut *a_tx) + .await + .expect("writer A locks the row"); + + // Writer B starts a guarded PATCH against its revision-0 snapshot on a + // SEPARATE pool. It enters `update_task`, issues its own FOR UPDATE, + // and blocks on A's lock. + let b_community = community; + let b_task_id = task.id; + let b_creator = creator.clone(); + let writer_b = tokio::spawn(async move { + update_task( + &b_pool, + b_community, + b_task_id, + &TaskPatch { + title: Some("writer B clobbers".to_owned()), + expected_revision: Some(task.revision), + ..TaskPatch::default() + }, + Some(&b_creator), + ) + .await + }); + + // Give writer B time to actually reach the blocked lock wait; without + // this the test can degrade into the sequential case by accident. + for _ in 0..50 { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + if writer_b.is_finished() { + break; + } + let blocked = sqlx::query_scalar::<_, i64>( + "SELECT count(*) FROM pg_stat_activity \ + WHERE wait_event_type = 'Lock' \ + AND query ILIKE '%tasks%' \ + AND pid <> pg_backend_pid()", + ) + .fetch_one(&a_pool) + .await + .unwrap_or(0); + if blocked > 0 { + break; + } + // B is still connecting or executing prior statements; keep polling. + } + + // Writer A commits its real change, releasing the lock: revision + // 0 -> 1 under the same history semantics as any other write. + sqlx::query(concat!( + "UPDATE tasks SET title = 'writer A won' ", + "WHERE community_id = $1 AND id = $2" + )) + .bind(community.as_uuid()) + .bind(task.id) + .execute(&mut *a_tx) + .await + .expect("writer A writes"); + a_tx.commit().await.expect("writer A commits"); + + // Writer B now acquires the lock, re-reads the row, and must see + // revision 1 against its expected 0 -> rejected, row untouched. + let b_result = writer_b.await.expect("writer B task panicked"); + match b_result { + Err(DbError::StaleRevision { + expected, actual, .. + }) => { + assert_eq!(expected, 0, "B's snapshot revision"); + assert_eq!(actual, 1, "A's committed revision"); + } + Ok(record) => panic!( + "interleaved stale write silently won (title now {:?})", + record.title + ), + other => panic!("expected StaleRevision, got {other:?}"), + } + + let after = get_task(&a_pool, community, task.id) + .await + .expect("re-fetch after the race"); + assert_eq!(after.title, "writer A won"); + assert_eq!(after.revision, 1); + + delete_test_community(&a_pool, community).await; + } + + /// HW-017: a body carrying only `expected_revision` asks for no change, so + /// `is_empty` must report it as empty. Otherwise it passes the relay's + /// "patch must change at least one field" gate and reaches the database as + /// a write that restates every column. + #[test] + fn a_guard_only_patch_is_empty() { + assert!( + TaskPatch { + expected_revision: Some(7), + ..TaskPatch::default() + } + .is_empty(), + "expected_revision is a precondition, not a requested change" + ); + assert!( + !TaskPatch { + title: Some("real".to_owned()), + expected_revision: Some(7), + ..TaskPatch::default() + } + .is_empty(), + "a guarded real change is not empty" + ); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn a_task_id_is_invisible_to_another_community() { diff --git a/crates/buzz-relay/src/api/tasks.rs b/crates/buzz-relay/src/api/tasks.rs index 030a3b91ca8..631a7201eb9 100644 --- a/crates/buzz-relay/src/api/tasks.rs +++ b/crates/buzz-relay/src/api/tasks.rs @@ -79,6 +79,9 @@ pub struct UpdateTaskRequest { due_at: Option>>, #[serde(default, deserialize_with = "deserialize_double_option")] assignee: Option>, + /// HW-017: optimistic concurrency guard. If present, the PATCH is rejected + /// with 409 when the task's `revision` does not match this value. + expected_revision: Option, } /// Body of `POST /api/tasks/{id}/events`. @@ -155,6 +158,11 @@ fn map_task_error(context: &str, error: buzz_db::DbError) -> (StatusCode, Json api_error(StatusCode::NOT_FOUND, "task not found"), buzz_db::DbError::InvalidData(message) => api_error(StatusCode::BAD_REQUEST, message), + buzz_db::DbError::StaleRevision { task_id, expected, actual } => { + api_error(StatusCode::CONFLICT, &format!( + "task {task_id} was modified by another writer: expected revision {expected}, found {actual}; re-fetch and retry" + )) + } buzz_db::DbError::AccessDenied(_) => api_error( StatusCode::SERVICE_UNAVAILABLE, "community writes are temporarily unavailable", @@ -437,6 +445,7 @@ pub async fn update_task( Some(None) => Some(None), Some(Some(raw)) => Some(Some(parse_pubkey("assignee", &raw)?)), }, + expected_revision: request.expected_revision, }; if patch.is_empty() { return Err(api_error( @@ -550,6 +559,7 @@ fn task_json(task: &TaskRecord) -> Value { "archived_at": task.archived_at.map(|value| value.timestamp()), "created_at": task.created_at.timestamp(), "updated_at": task.updated_at.timestamp(), + "revision": task.revision, }) } @@ -686,6 +696,7 @@ mod tests { archived_at: None, created_at: Utc::now(), updated_at: Utc::now(), + revision: 0, }; let wire = task_json(&task); assert_eq!(wire["status"], "in_progress"); diff --git a/desktop/src-tauri/src/commands/tasks.rs b/desktop/src-tauri/src/commands/tasks.rs index 2dfcc5c213b..acde48d7c25 100644 --- a/desktop/src-tauri/src/commands/tasks.rs +++ b/desktop/src-tauri/src/commands/tasks.rs @@ -55,6 +55,7 @@ pub struct ChannelTask { pub done_at: Option, pub created_at: i64, pub updated_at: i64, + pub revision: i64, } impl ChannelTask { @@ -75,6 +76,7 @@ impl ChannelTask { done_at: value["done_at"].as_i64(), created_at: value["created_at"].as_i64().unwrap_or_default(), updated_at: value["updated_at"].as_i64().unwrap_or_default(), + revision: value["revision"].as_i64().unwrap_or_default(), } } } @@ -223,9 +225,13 @@ pub async fn tasks_set_status( state: State<'_, AppState>, task_id: String, status: String, + expected_revision: Option, ) -> Result { let path = format!("{TASKS_PATH}/{task_id}"); - let payload = serde_json::json!({ "status": status }); + let mut payload = serde_json::json!({ "status": status }); + if let Some(rev) = expected_revision { + payload["expected_revision"] = serde_json::json!(rev); + } let value = tasks_request( state.inner(), reqwest::Method::PATCH, @@ -254,10 +260,14 @@ pub async fn tasks_set_assignee( state: State<'_, AppState>, task_id: String, assignee: Option, + expected_revision: Option, ) -> Result { let path = format!("{TASKS_PATH}/{task_id}"); // serde_json::Value::Null is emitted for `None` — the unassign case. - let payload = serde_json::json!({ "assignee": assignee }); + let mut payload = serde_json::json!({ "assignee": assignee }); + if let Some(rev) = expected_revision { + payload["expected_revision"] = serde_json::json!(rev); + } let value = tasks_request( state.inner(), reqwest::Method::PATCH, diff --git a/desktop/src/features/tasks/lib/channelTasks.ts b/desktop/src/features/tasks/lib/channelTasks.ts index 1751af4e124..ee91a9f7354 100644 --- a/desktop/src/features/tasks/lib/channelTasks.ts +++ b/desktop/src/features/tasks/lib/channelTasks.ts @@ -23,6 +23,8 @@ export type ChannelTask = { doneAt: number | null; createdAt: number; updatedAt: number; + /** HW-017: monotonic revision counter for optimistic concurrency. */ + revision: number; }; /** One source community's outcome in the My-Tasks fan-in. */ diff --git a/migrations/0050_task_optimistic_concurrency.sql b/migrations/0050_task_optimistic_concurrency.sql new file mode 100644 index 00000000000..cdadb44d598 --- /dev/null +++ b/migrations/0050_task_optimistic_concurrency.sql @@ -0,0 +1,62 @@ +-- HW-017: Optimistic concurrency guard for task PATCH. +-- +-- A stale write (two clients fetch the same task, both PATCH, the second +-- clobbering the first) must not silently win. `revision` is a monotonic +-- counter that increments on every UPDATE, so a PATCH carrying +-- `expected_revision` can compare-and-swap: if the row's revision does not +-- match, the relay returns 409 and the caller re-fetches. +-- +-- The counter is maintained by a BEFORE UPDATE trigger so it applies to every +-- write path, not just the relay's `update_task`. This keeps the guarantee +-- structural rather than convention-based. The trigger bumps only when the +-- row's payload actually changed, so an idempotent restate cannot manufacture +-- a spurious conflict for other readers (see the function body). +-- +-- Backward compatibility: a PATCH that omits `expected_revision` (every +-- existing client at the time this ships) skips the guard entirely and +-- behaves exactly as before. The column defaults to 0 so existing rows +-- receive a revision without a backfill; the trigger sets it to 1 on the +-- first post-migration UPDATE that changes something. +-- +-- Additive only: ALTER TABLE ... ADD COLUMN does not rewrite history that +-- brownfield relays have already applied, so 0001's checksum is untouched. +SET LOCAL lock_timeout = '5s'; + +ALTER TABLE tasks ADD COLUMN revision INT NOT NULL DEFAULT 0; + +-- A monotonic counter the relay can compare-and-swap against. The trigger +-- fires on every UPDATE, so the guarantee is structural — not a convention +-- the relay could forget. +CREATE OR REPLACE FUNCTION bump_task_revision() +RETURNS TRIGGER AS $$ +BEGIN + -- Bump ONLY when the row's payload actually changed. + -- + -- A statement that restates every column at its existing value (a client + -- retry, or a PATCH that sets status to the status it already holds) still + -- fires a BEFORE UPDATE trigger. Bumping there would be a correctness bug, + -- not a harmless extra: it would invalidate every other client's + -- `expected_revision` for a write that changed nothing, so an idempotent + -- retry would manufacture spurious 409s. The existing task-event logic + -- already suppresses same-value events (`status_change_action` returns + -- None on an unchanged status); revision must agree with that guarantee. + -- + -- The two derived columns are first normalised to their OLD values so the + -- whole-row comparison sees only caller-supplied payload. Comparing the + -- entire row rather than an enumerated column list means a future column + -- added to `tasks` is covered automatically — an explicit list would + -- silently stop guarding whatever someone forgot to add to it. + NEW.revision := OLD.revision; + NEW.updated_at := OLD.updated_at; + IF NEW IS DISTINCT FROM OLD THEN + NEW.revision := OLD.revision + 1; + NEW.updated_at := NOW(); + END IF; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +CREATE TRIGGER trg_tasks_revision + BEFORE UPDATE ON tasks + FOR EACH ROW + EXECUTE FUNCTION bump_task_revision(); diff --git a/mobile/lib/shared/tasks/task.dart b/mobile/lib/shared/tasks/task.dart index 23086772fed..5ce874cbe8a 100644 --- a/mobile/lib/shared/tasks/task.dart +++ b/mobile/lib/shared/tasks/task.dart @@ -102,6 +102,7 @@ class Task { required this.priority, required this.createdAt, required this.updatedAt, + required this.revision, this.channelId, this.createdBy, this.assignee, @@ -128,6 +129,7 @@ class Task { priority: json['priority'] is int ? json['priority'] as int : 0, createdAt: _dateFromSeconds(json['created_at']) ?? DateTime.now().toUtc(), updatedAt: _dateFromSeconds(json['updated_at']) ?? DateTime.now().toUtc(), + revision: json['revision'] is int ? json['revision'] as int : 0, channelId: _stringOrNull(json['channel_id']), createdBy: _stringOrNull(json['created_by']), assignee: _stringOrNull(json['assignee']), @@ -159,6 +161,9 @@ class Task { /// When the task last changed. final DateTime updatedAt; + /// Monotonic revision counter for optimistic concurrency (HW-017). + final int revision; + /// Channel this task is scoped to, or null for a community-wide task. final String? channelId; diff --git a/mobile/lib/shared/tasks/tasks_api.dart b/mobile/lib/shared/tasks/tasks_api.dart index 5ff9a012850..895af5a9372 100644 --- a/mobile/lib/shared/tasks/tasks_api.dart +++ b/mobile/lib/shared/tasks/tasks_api.dart @@ -144,11 +144,13 @@ class TasksApi { TaskStatus? status, String? title, int? priority, + int? expectedRevision, }) async { final payload = { if (status != null) 'status': status.wireValue, if (title != null) 'title': title.trim(), 'priority': ?priority, + 'expected_revision': ?expectedRevision, }; final decoded = await _send('PATCH', _uri('/api/tasks/$taskId'), payload); return Task.fromJson(_asObject(decoded)); diff --git a/mobile/test/features/channels/thread_task_chip_test.dart b/mobile/test/features/channels/thread_task_chip_test.dart index fdb989d7888..6a4855fae95 100644 --- a/mobile/test/features/channels/thread_task_chip_test.dart +++ b/mobile/test/features/channels/thread_task_chip_test.dart @@ -13,6 +13,7 @@ Task _task({ }) => Task( id: id, title: title, + revision: 0, status: status, priority: 0, createdAt: DateTime.fromMillisecondsSinceEpoch( diff --git a/schema/schema.sql b/schema/schema.sql index 2a27512f5c0..1944d5cc570 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -1254,6 +1254,7 @@ CREATE TABLE tasks ( archived_at TIMESTAMPTZ, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + revision INT NOT NULL DEFAULT 0, PRIMARY KEY (community_id, id), CONSTRAINT chk_tasks_done_at_matches_status CHECK ((status = 'done') = (done_at IS NOT NULL)), @@ -1281,6 +1282,32 @@ CREATE INDEX idx_tasks_community_channel ON tasks (community_id, channel_id) CREATE INDEX idx_tasks_community_parent ON tasks (community_id, parent_task_id) WHERE parent_task_id IS NOT NULL; +-- HW-017: monotonic revision counter for optimistic concurrency on PATCH. +CREATE OR REPLACE FUNCTION bump_task_revision() +RETURNS TRIGGER AS $$ +BEGIN + -- Bump ONLY when the row's payload actually changed. An idempotent + -- restate still fires a BEFORE UPDATE trigger; bumping there would + -- invalidate every other client's `expected_revision` for a write that + -- changed nothing, manufacturing spurious 409s. The derived columns are + -- normalised to OLD first so the whole-row comparison sees only + -- caller-supplied payload, and comparing the whole row means a future + -- column on `tasks` is guarded automatically. + NEW.revision := OLD.revision; + NEW.updated_at := OLD.updated_at; + IF NEW IS DISTINCT FROM OLD THEN + NEW.revision := OLD.revision + 1; + NEW.updated_at := NOW(); + END IF; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +CREATE TRIGGER trg_tasks_revision + BEFORE UPDATE ON tasks + FOR EACH ROW + EXECUTE FUNCTION bump_task_revision(); + -- Append-only lifecycle and comment log; also the read model behind the -- human-visible task feed, hence the (community, time) feed index. CREATE TABLE task_events (