Compare and set the tick size writer and group release on the ledger server

changelog.d/9612.fixed.md

@@ -0,0 +1 @@
+- The tick size writer and group release send the fields they read as the ledger guard, so the server compares and sets them and a concurrent size or regroup is never overwritten; `task_ungroup` accepts an `if_match` of `group_members` and refuses the release when they moved.

mutation-clearances/3dbb7d859eb64595f08e59108eea92e71e420d291a2d64223c95256de2da2ace.json

@@ -0,0 +1,6 @@
+{
+  "scripts/tick/steps/tasks.py:scripts.tick.steps.tasks.x__write_task:428a32bb257ad3d52cbf60ac5dc0383b59b294ecff53cafe5f13881e2d399958": {
+    "reader": "standards_reader",
+    "reason": "typing.cast returns its second argument unchanged and ignores the type argument at runtime, so cast(None, write.value) and cast(dict[str, Any], write.value) pass the same mapping to update_task; the mutant changes only a type annotation and cannot change any output."
+  }
+}

scripts/swarm/ledger_client.py

@@ -277,8 +277,9 @@ def freeze(
     def group_tasks(self, slug, lead, members):
         self._call(slug, [_op("task_group", "swarm", item=f"tasks/{lead}", members=list(members))])
 
-    def ungroup_tasks(self, slug, lead):
-        self._call(slug, [_op("task_ungroup", "swarm", item=f"tasks/{lead}")])
+    def ungroup_tasks(self, slug: str, lead: str, if_match: dict[str, list[str]] | None = None) -> None:
+        guard = {} if if_match is None else {"if_match": if_match}
+        self._call(slug, [_op("task_ungroup", "swarm", item=f"tasks/{lead}", **guard)])
 
     def rank_task(self, slug, task_id, rank, by, if_owned=False):
         guard = {"if_owned": True} if if_owned else {}

scripts/swarm_ledger/api/schemas.py

@@ -89,7 +89,7 @@ def validate(schema: dict, body: object) -> None:
     "task_update": "by item fields if_state if_match if_plan_lines_missing quote",
     "task_rank": "by item rank if_owned",
     "task_group": "by item members",
-    "task_ungroup": "by item",
+    "task_ungroup": "by item if_match",
     "task_split": "by item halves",
     "title_set": "text",
     "overview_set": "text",

scripts/swarm_ledger/ledger_groups.py

@@ -4,6 +4,7 @@
 """
 
 import re
+from typing import Any
 
 from scripts.swarm_ledger import ledger_kinds, ports
 
@@ -19,13 +20,17 @@
 ITEM_RE = re.compile(r"^tasks/[^/]+$")
 
 
-def check(op):
+def check(op: dict[str, Any]) -> None:
     by = op.get("by")
     if not isinstance(by, str) or not AUTHOR_RE.match(by) or by == "operator":
         raise ValueError(f"{op['op']} needs `by`, an agent name other than operator")
     if op["op"] == "task_ungroup":
-        if set(op) != {"op", "id", "by", "item"} or not ITEM_RE.match(str(op["item"])):
-            raise ValueError("task_ungroup takes only an id, by and an item tasks/<lead id>")
+        if set(op) - {"if_match"} != {"op", "id", "by", "item"} or not ITEM_RE.match(str(op["item"])):
+            raise ValueError("task_ungroup takes only an id, by, an item tasks/<lead id> and an if_match")
+        expected = op.get("if_match", {"group_members": []})
+        members = expected.get("group_members") if isinstance(expected, dict) else None
+        if not isinstance(members, list) or set(expected) != {"group_members"} or not _task_ids(members):
+            raise ValueError("task_ungroup if_match holds only group_members, a list of task ids")
         return
     if set(op) != {"op", "id", "by", "item", "members"} or not ITEM_RE.match(str(op["item"])):
         raise ValueError("task_group takes only an id, by, an item tasks/<lead id> and members")
@@ -42,6 +47,11 @@ def check(op):
         raise ValueError(f"a group holds at most {MAX_TASKS} tasks")
 
 
+def _task_ids(members: list[Any]) -> bool:
+    """Whether every member is a task id."""
+    return all(isinstance(m, str) and ID_RE.match(m) for m in members)
+
+
 def refusal(tasks: list[dict], known: dict, ceiling: int = CEILING) -> str:
     """Why these tasks cannot form one group, or '' when they can."""
     if len({(t.get("lane"), t.get("profile") or "", ledger_kinds.kind(t)) for t in tasks}) > 1:
@@ -104,11 +114,14 @@ def _group(known, op, ctx):
     return True
 
 
-def _ungroup(known, op, ctx):
+def _ungroup(known: dict[str, dict[str, Any]], op: dict[str, Any], ctx: Any) -> bool:
     lead_id = op["item"].split("/")[1]
     if not known.get(lead_id, {}).get("group_members"):
         ctx.refused.append(f"{op['item']} leads no group")
         return False
+    if "if_match" in op and known[lead_id]["group_members"] != op["if_match"]["group_members"]:
+        ctx.refused.append(f"{op['item']} changed its group members")
+        return False
     released = [m for m in known[lead_id].pop("group_members") if known.get(m, {}).get("merged_into") == lead_id]
     ctx.stamp(f"{op['item']}/group_members", op["by"])
     for member in released:

scripts/tick/steps/tasks.py

@@ -293,16 +293,22 @@ def task_writer(slug: str, ledger: Any, read: Callable[[str], dict[str, Any] | N
 
 
 def _write_task(slug: str, ledger: Any, read: Callable[[str], dict[str, Any] | None], write: Write) -> None:
-    """Write the task fields of ``write`` unless a field it expects moved or the ledger refuses it."""
+    """Write the task fields of ``write`` unless a field it expects moved or the ledger refuses it.
+
+    The expected fields go to the ledger as its ``if_match`` guard, so the server compares and sets them in one apply.
+    """
     expected = cast(dict[str, Any], write.expected or {})
     task = read(write.key)
     moved = [] if task is None else sorted(key for key, value in expected.items() if task.get(key, "") != value)
     if moved:
         raise RevisionConflict(f"task {write.key} changed {', '.join(moved)} since the step read it")
+    fields = cast(dict[str, Any], write.value)
     try:
-        ledger.update_task(slug, write.key, write.value)
+        held = ledger.update_task(slug, write.key, fields, if_match=expected or None)
     except LedgerRefused as exc:
         raise RevisionConflict(f"the ledger refused the write of task {write.key}") from exc
+    if changed := sorted(key for key in expected if key in fields and held.get(key, "") != fields[key]):
+        raise RevisionConflict(f"task {write.key} changed {', '.join(changed)} on the ledger before the write")
 
 
 def appliers(state: dict[str, Any]) -> dict[Domain, Callable[[Write], list[str] | None]]:
@@ -355,13 +361,16 @@ def group_writer(slug: str, ledger: Any, read: Callable[[str], dict[str, Any] |
 
 
 def _release_group(slug: str, ledger: Any, read: Callable[[str], dict[str, Any] | None], write: Write) -> None:
-    """Release the group under the lead of ``write`` unless its members moved or the ledger refuses it."""
+    """Release the group under the lead of ``write`` unless its members moved or the ledger refuses it.
+
+    The expected members go to the ledger as its ``if_match`` guard, so the server compares them before it releases.
+    """
     expected = cast(dict[str, Any], write.expected or {})
     task = read(write.key)
     if task is not None and list(task.get("group_members") or []) != expected.get("group_members", []):
         raise RevisionConflict(f"task {write.key} changed group_members since the step read it")
     try:
-        ledger.ungroup_tasks(slug, write.key)
+        ledger.ungroup_tasks(slug, write.key, if_match={"group_members": list(expected.get("group_members", []))})
     except LedgerRefused as exc:
         raise RevisionConflict(f"the ledger refused the release of the group under task {write.key}") from exc
 

tests/TYPE_BASELINE.json

@@ -621,7 +621,7 @@
   "scripts/swarm/launch_job.py": 211,
   "scripts/swarm/launch_model.py": 53,
   "scripts/swarm/lease.py": 47,
-  "scripts/swarm/ledger_client.py": 433,
+  "scripts/swarm/ledger_client.py": 428,
   "scripts/swarm/ledger_cutover.py": 83,
   "scripts/swarm/ledger_events.py": 1122,
   "scripts/swarm/ledger_host.py": 27,
@@ -794,7 +794,7 @@
   "scripts/swarm_ledger/ledger_fold.py": 269,
   "scripts/swarm_ledger/ledger_freezes.py": 83,
   "scripts/swarm_ledger/ledger_gate.py": 173,
-  "scripts/swarm_ledger/ledger_groups.py": 119,
+  "scripts/swarm_ledger/ledger_groups.py": 92,
   "scripts/swarm_ledger/ledger_health.py": 11,
   "scripts/swarm_ledger/ledger_hook.py": 267,
   "scripts/swarm_ledger/ledger_kinds.py": 58,

tests/swarm/test_grouping.py

@@ -415,7 +415,7 @@ def state(self, slug):
         found["_meta"]["stamps"] = self.stamps
         return found
 
-    def ungroup_tasks(self, slug: str, lead: str) -> None:
+    def ungroup_tasks(self, slug: str, lead: str, if_match: dict[str, list[str]] | None = None) -> None:
         assert slug == "sw"
         self.ungrouped.append(lead)
         if lead in self.view:

tests/swarm/test_ledger_client_group.py

@@ -1,5 +1,7 @@
 from types import SimpleNamespace
 
+import pytest
+
 from scripts.swarm import ledger_client
 
 
@@ -36,3 +38,16 @@ def call(slug, batch, service):
     assert (slug, service) == ("demo", True)
     assert op["id"].startswith("task_ungroup-")
     assert {k: v for k, v in op.items() if k != "id"} == {"op": "task_ungroup", "by": "swarm", "item": "tasks/t1"}
+
+
+def test_ungroup_tasks_sends_the_expected_members_as_its_guard(monkeypatch: pytest.MonkeyPatch) -> None:
+    sent: list[dict[str, object]] = []
+
+    def call(slug: str, batch: list[dict[str, object]], service: bool) -> dict[str, list[str]]:
+        sent.extend(batch)
+        return {"rejected": []}
+
+    monkeypatch.setattr(ledger_client, "_ledger", lambda: SimpleNamespace(call=call))
+    ledger_client.LedgerClient().ungroup_tasks("demo", "t1", if_match={"group_members": ["t2"]})
+    [op] = sent
+    assert op["if_match"] == {"group_members": ["t2"]}