feat: bind job events to command leases

2026-08-13 06:45:54 +00:00
parent 4ec852014d
commit 03b6751d42
12 changed files with 71 additions and 17 deletions
@@ -2,7 +2,7 @@

 Source-of-truth protobuf contract for Archive Control clients and the Titan
 control daemon. Implementations exchange canonical protobuf JSON as one text
-message per WebSocket frame.
+message per WebSocket frame. The current incompatible protocol major is `2`.

 Upstream: `cabbage/archive-control-proto`

@@ -29,14 +29,16 @@ Upstream: `cabbage/archive-control-proto`
 3. A server command is delivered at least once. Clients durably deduplicate
   `command_id` and return `DUPLICATE` for a replayed accepted command.
 4. `CommandAck` means durable acceptance, not operation completion.
-5. Job events have monotonically increasing global per-job sequences. A command
-   carries the accepted sequence base and grants one client the event-writer
-   lease; a gap requires `JobSnapshot` reconciliation before further
-   destructive transitions.
+5. Job events have monotonically increasing global per-job sequences and name
+   the durable command that owns them. Control verifies the reporting client,
+   command cursor, and permitted state transition. A gap or conflict requires
+   authoritative `ReconcileJobCommand` processing before further destructive
+   transitions.
 6. UUIDs use lowercase canonical text. Info hashes use validated lowercase hex.
 7. `SelectionSet.ranges` are inclusive, sorted, non-overlapping, and coalesced.
-8. Receivers ignore unknown JSON fields for compatible minor versions. They
-   reject an unknown required command/job operation.
+8. Receivers reject a different protocol major. Within major `2`, they ignore
+   unknown JSON fields only for compatible minor versions and reject an unknown
+   required command/job operation.
 9. Published field numbers and enum values are never reused. Removed values are
   reserved in the change that removes them.
 10. Implementations cap an envelope at the negotiated maximum and keep streamed
@@ -1,6 +1,6 @@
 {
  "protocolVersion": {
-    "major": 1,
+    "major": 2,
    "minor": 0
  },
  "messageId": "2fc6a524-84cd-47ab-bd73-30d463b7c43f",
@@ -1,6 +1,6 @@
 {
  "protocolVersion": {
-    "major": 1,
+    "major": 2,
    "minor": 0
  },
  "messageId": "9c62fcda-5f83-4286-80cf-a628001e44d3",
@@ -1,6 +1,6 @@
 {
  "protocolVersion": {
-    "major": 1,
+    "major": 2,
    "minor": 0
  },
  "messageId": "7c4a3534-c754-4c4d-a12e-1a1047e3f96e",
@@ -1,6 +1,6 @@
 {
  "protocolVersion": {
-    "major": 1,
+    "major": 2,
    "minor": 0
  },
  "messageId": "56aae470-d1c9-402a-bf14-d7a28ff8998f",
@@ -1,6 +1,6 @@
 {
  "protocolVersion": {
-    "major": 1,
+    "major": 2,
    "minor": 0
  },
  "messageId": "830626f3-dc76-46f0-a9b1-a6fb61b3ba73",
@@ -1,6 +1,6 @@
 {
  "protocolVersion": {
-    "major": 1,
+    "major": 2,
    "minor": 0
  },
  "messageId": "d00def27-cde1-43b7-a6af-ee3582d0c7c1",
@@ -1,6 +1,6 @@
 {
  "protocolVersion": {
-    "major": 1,
+    "major": 2,
    "minor": 0
  },
  "messageId": "f337dc9c-84b8-42c6-a2ae-52f8f91258a8",
@@ -8,6 +8,7 @@
  "sentAt": "2026-07-22T12:08:00Z",
  "jobEvent": {
    "eventId": "ecfbd79e-e0fe-41da-ab77-012cde0ed53f",
+    "commandId": "6bfccbf4-866f-46fe-bf67-d8943c1ece91",
    "jobId": "fd6844d0-ed76-475e-8375-2f98d8eea470",
    "sequence": "23",
    "jobRevision": "9",
@@ -1,6 +1,6 @@
 {
  "protocolVersion": {
-    "major": 1,
+    "major": 2,
    "minor": 0
  },
  "messageId": "44b64096-cd58-4f49-bae4-d829f50f787c",
@@ -8,6 +8,7 @@
  "sentAt": "2026-07-22T12:04:00Z",
  "jobEvent": {
    "eventId": "18cf6a34-b0a5-429e-b56a-9228b2dca640",
+    "commandId": "6bfccbf4-866f-46fe-bf67-d8943c1ece91",
    "jobId": "fd6844d0-ed76-475e-8375-2f98d8eea470",
    "sequence": "17",
    "jobRevision": "6",
@@ -0,0 +1,34 @@
+{
+  "protocolVersion": {
+    "major": 2
+  },
+  "messageId": "4be5df55-3d95-4f60-8f6b-1cfd0b64cfbb",
+  "sentAt": "2026-08-13T06:00:00Z",
+  "command": {
+    "commandId": "94b4f73f-31a8-426b-8511-12b91987f1b4",
+    "createdAt": "2026-08-13T06:00:00Z",
+    "reconcileJob": {
+      "authoritativeJob": {
+        "definition": {
+          "jobId": "3173ae0d-ecc4-4113-a659-7897aac6c66f",
+          "idempotencyKey": "55b6d968-e20d-444b-8fae-17914d15fdf5",
+          "operation": "JOB_OPERATION_ARCHIVE",
+          "resourceDisplayName": "example",
+          "createdAt": "2026-08-13T05:00:00Z",
+          "transfer": {
+            "sourceClientId": "cache-1",
+            "targetClientId": "archive-1",
+            "routeId": "route-example"
+          }
+        },
+        "state": "JOB_STATE_RUNNING",
+        "revision": "12"
+      },
+      "authoritativeLastEventSequence": "12",
+      "supersededCommandIds": [
+        "6d4ab7f8-2ff0-47d8-8c7c-45a67dce8535"
+      ],
+      "reason": "control cursor is authoritative after stale event replay"
+    }
+  }
+}
@@ -1,13 +1,13 @@
 {
  "protocolVersion": {
-    "major": 1,
+    "major": 2,
    "minor": 0
  },
  "messageId": "0d6d31f8-1242-4a40-a261-433897102fe4",
  "sentAt": "2026-07-22T12:00:00Z",
  "registerRequest": {
    "protocolVersion": {
-      "major": 1,
+      "major": 2,
      "minor": 0
    },
    "client": {
@@ -38,6 +38,17 @@ message RequestJobSnapshotCommand {
  repeated string job_ids = 1;
 }

+// ReconcileJobCommand makes control's durable cursor authoritative after a
+// reconnect conflict, local database restore, or rejected stale event. The
+// client must durably adopt this cursor before acknowledging the command and
+// must never emit output for superseded_command_ids afterwards.
+message ReconcileJobCommand {
+  JobRecord authoritative_job = 1;
+  uint64 authoritative_last_event_sequence = 2;
+  repeated string superseded_command_ids = 3;
+  string reason = 4;
+}
+
 message Command {
  string command_id = 1;
  google.protobuf.Timestamp created_at = 2;
@@ -48,6 +59,7 @@ message Command {
    EnsureRouteCommand ensure_route = 13;
    InventoryQuery inventory_query = 14;
    RequestJobSnapshotCommand request_job_snapshot = 15;
+    ReconcileJobCommand reconcile_job = 16;
  }
 }

@@ -97,6 +109,10 @@ message JobEvent {
  ResourceStateFingerprint observed_resource = 10;
  Placement observed_placement = 11;
  google.protobuf.Timestamp occurred_at = 12;
+  // The durable ExecuteStepCommand or CancelJobCommand that owns this event.
+  // Control verifies this ID, the reporting client, the command's expected
+  // cursor, and the permitted step/state transition before appending it.
+  string command_id = 13;
 }

 message JobSnapshot {