Guard value reads, hold back emptied places, run once an hour

.github/workflows/sync.yml

@@ -2,7 +2,7 @@ name: Board sync
 
 on:
   schedule:
-    - cron: "7,37 * * * *"
+    - cron: "11 * * * *"
   workflow_dispatch:
     inputs:
       apply:
@@ -65,6 +65,7 @@ jobs:
           BOARD_SYNC_TOKEN_INTERRA_PROJECT: ${{ steps.interra.outputs.token }}
           BOARD_SYNC_TOKEN_FORMINGWORLDS: ${{ steps.formingworlds.outputs.token }}
           APPLY: ${{ github.event_name == 'schedule' || inputs.apply }}
+          EVENT: ${{ github.event_name }}
           BULK: ${{ github.event_name == 'workflow_dispatch' && inputs.allow_bulk }}
         run: |
           flags="--state state/state.json --snapshot-dir state --report state/report.txt --summary --pause 1"
@@ -76,6 +77,12 @@ jobs:
             echo "::warning::Some writes are parked after repeated failures; see report.txt in the state repository."
             code=0
           fi
+          if [ "$code" = 5 ] && [ "$EVENT" = schedule ]; then
+            echo "::notice::Skipped: too soon after the last run that wrote; see the log above."
+            code=0
+          elif [ "$code" = 5 ]; then
+            echo "::error::Nothing done: too soon after the last run that wrote (see the log above); start this run again later."
+          fi
           exit "$code"
 
       - name: Store state, snapshot and report

README.md

@@ -31,18 +31,25 @@ Effort follows Workload in one direction: a hand edit of Effort on an issue that
   A board does not record when a value was cleared, so a cleared board value loses to a value set at another place in the same interval.
   The issue timeline read covers the last 100 field events; an older change has no time and loses.
 - A value that every changed place already shows, after the name map, wins without a conflict.
-- A cleared value is cleared in the other places too.
+- A cleared value is cleared in the other places too, except where a place is held back (see below).
   A field that is emptied as a whole at one place (all of its 5 or more values, see below) is filled again from the others.
 - A field that is missing at a place is skipped there, never read as empty, and the run log names it; a value without a matching option is listed in `report.txt` as skipped.
   A field with a new id (deleted and created again), or a field that held 5 or more values and now holds none, does not clear the other places.
 - An issue that is not on a board, or cannot be read, keeps its record for that board.
-  When it is there again with the field, its empty fields receive the agreed values, an unchanged value counts as no change, and a different value wins only when it is newer than the agreed value (an agreed value of unknown time keeps its place).
+  When it is there again with the field, its empty fields receive the agreed values, an unchanged value counts as no change, and a different value wins only when it is newer than the agreed value (an agreed value of unknown time keeps its place); an issue field that was cleared after the agreed value counts as a clear, since its removal event has a time.
   On a field that was created again, the first value counts as a change, as on a new board.
   An issue that is on no board for 90 days is dropped from the state.
 - A list longer than one page of the API (fields of a board, values of an item, issue fields, issue field values) stops the read with an error, so a cut-off list is never read as empty values.
-- A board read is short when it finds no items, a number of items other than the board's total, or a total that changes between pages, since the API can return an empty page without an error.
+- A board read is short when it finds no items, a number of items other than the board's total, a total that changes between pages, or an issue item with no values or a number of values other than its total (every item holds at least its title), since the API can return an empty page without an error.
   The board is then read again, up to 6 reads in all, 10 seconds apart, and the run stops with exit code 2 before any write when all 6 are short.
   So a board with no items stops every run, and the retries count against the 10 minutes of a run.
+  A read of issue field values is short when it returns fewer issues than asked for, or an issue with a number of values other than its total; it is read again in the same way.
+  An issue that lacks the value of a synced field whose latest event set it counts as unread for its issue fields.
+- A board place of an issue whose synced fields held 2 or more values in the state and all read empty now is held back for the run, as if the place did not hold the issue, and `report.txt` lists it; fields that were created again or emptied as a whole do not count.
+  The issue fields of an issue are held back in the same way when one synced field that held a value reads empty without a removal event, since a clear of an issue field always has one.
+  So a read that drops all values of a board item that held 2 or more, or a value of an issue field, clears nothing; when the place reads the same in the next run, it receives the agreed values.
+  A board item with 1 synced value that reads empty, or a read that drops some but not all values of a board item, is not caught: it counts as a clear.
+  A hand clear of all synced fields of an issue on one board is therefore undone; clear them one by one in different runs, or on the issue.
 - A run that plans more than 20 writes stops and writes nothing (exit code 3).
   Read the plan in `report.txt` in the state repository, then start the workflow by hand with `apply` and `allow_bulk`.
   The state then holds the approved writes that are not done yet, and the scheduled runs write them, at most 400 values per run.
@@ -58,7 +65,10 @@ Effort follows Workload in one direction: a hand edit of Effort on an issue that
   Issue links, values and error text stay in the state repository.
 - A run reads the places one after the other and then writes.
   An edit made in the minutes of a run can be overwritten by that run; the next run then sees the written value as agreed.
-- Exit codes: 0 done, 1 some writes failed, 2 the read, the state file, a token or the run failed, 3 too many changes, 4 done but some writes are parked (the workflow shows it as a warning), 143 or 130 stopped by a signal.
+- A run with `apply` that starts less than 50 minutes after the start of the last run that tried a write stops before the read with exit code 5 (`--min-interval`, 0 turns the check off); a run without `apply` is not limited.
+  A scheduled run that stops this way shows a notice; a run started by hand fails, so start it again later.
+  A stored start time more than 5 minutes later than the run's own start is an error (exit code 2) on every run with `apply`; to recover, delete `applied_at` from `state.json` in the state repository.
+- Exit codes: 0 done, 1 some writes failed, 2 the read, the state file, a token or the run failed, 3 too many changes, 4 done but some writes are parked (the workflow shows it as a warning), 5 too soon after the last run that wrote, 143 or 130 stopped by a signal.
 
 ## Setup
 
@@ -72,7 +82,7 @@ Effort follows Workload in one direction: a hand edit of Effort on an issue that
 4. Create `Interra-Project/board-sync-state` as a private repository with one commit on `main`, for example a README.
 5. Store all 3 values in the environment only, not as repository secrets: the App client id as the variable `BOARD_SYNC_CLIENT_ID`, the App private key as the secret `BOARD_SYNC_PRIVATE_KEY`, and the private key of an SSH deploy key with write access to `Interra-Project/board-sync-state` as the secret `STATE_DEPLOY_KEY`.
 
-The workflow runs at minute 7 and 37 of every hour on `main`; a run on another branch is skipped.
+The workflow runs at minute 11 of every hour on `main`; a run on another branch is skipped.
 A run started by hand from the Actions tab only plans, unless `apply` is set.
 GitHub turns off scheduled workflows in a public repository after 60 days without activity in that repository.
 Check the date of the last run in the Actions tab of this repository, and enable the workflow again there when it is off.

board_sync/__main__.py

@@ -5,7 +5,9 @@
 writes the plan, unless the plan has more than ``--max-changes`` writes; then
 it stops with exit code 3. A person reads the plan and reruns with
 ``--allow-bulk``; the state then holds the approved writes, and the next runs
-write them in batches of ``--limit``.
+write them in batches of ``--limit``. A run with ``--apply`` that starts less
+than ``--min-interval`` minutes after the start of the last run that wrote
+stops before the read, with exit code 5.
 """
 
 import argparse
@@ -164,7 +166,7 @@ def apply(args, clients, snap, result, active, parked, state, now, summary, deta
     """
     old_fails = state.get("failures", {})
     batch, later = active[: args.limit], active[args.limit :]
-    failed, done_keys, done, in_row, late = [], set(), 0, 0, False
+    failed, done_keys, done, in_row, late, tried = [], set(), 0, 0, False, False
     try:
         for n, w in enumerate(batch):
             if in_row >= STOP_AFTER_FAILURES:
@@ -174,6 +176,7 @@ def apply(args, clients, snap, result, active, parked, state, now, summary, deta
                 break
             if n and args.pause:
                 time.sleep(args.pause)
+            tried = True
             try:
                 github.write(clients, snap, w)
                 done_keys.add(write_key(w))
@@ -201,6 +204,10 @@ def apply(args, clients, snap, result, active, parked, state, now, summary, deta
         idle = 0 if progress else (0 if args.allow_bulk else state.get("bulk_idle", 0)) + 1
         keep = left and idle < BULK_IDLE_RUNS
         result.state["bulk_keys"], result.state["bulk_idle"] = (left, idle) if keep else ([], 0)
+        # Only a run that tried a write moves the start of the interval.
+        last = args.started_at.isoformat(timespec="seconds") if tried else state.get("applied_at")
+        if last:
+            result.state["applied_at"] = last
         save_state(args.state, result.state)
     if in_row >= STOP_AFTER_FAILURES:
         summary.append(f"stopped after {in_row} failed writes in a row")
@@ -240,6 +247,12 @@ def plan_and_apply(args, clients, snap, out):
     summary += [f"reset: {key} / {fkey} is treated as new, its old values are ignored" for key, fkey in result.resets]
     if result.skipped:
         summary.append(f"skipped: {len(result.skipped)} values without an option at the place")
+    if result.held:
+        summary.append(f"held back: {len(result.held)} places of an issue with values missing, listed in the report")
+        detail += [
+            f"held back {snap['items'][i]['url']} {config.BOARDS.get(p, {}).get('title', 'issue field')}: values missing from the read"
+            for i, p in result.held
+        ]
     if parked:
         summary.append(f"parked: {len(parked)} writes that failed in {PARK_AFTER} runs, listed in the report")
     approved_keys = set(state.get("bulk_keys", []))
@@ -257,6 +270,29 @@ def plan_and_apply(args, clients, snap, out):
     return code or (4 if parked and args.apply else 0)
 
 
+# A stored start time later than this run's start by more than this is a damaged state.
+CLOCK_SLACK = timedelta(minutes=5)
+
+
+def too_soon(args):
+    """Return the start time of the last run that wrote, when it is less than ``args.min_interval`` minutes ago.
+
+    Raises
+    ------
+    StateError
+        If the stored time is more than ``CLOCK_SLACK`` later than this run's start.
+    """
+    if not args.apply or args.min_interval <= 0:
+        return None
+    last = load_state(args.state).get("applied_at")
+    if not last:
+        return None
+    gap = args.started_at - datetime.fromisoformat(last)
+    if gap < -CLOCK_SLACK:
+        raise StateError(f"applied_at: the last run that wrote started at {last}, after this run")
+    return last if gap < timedelta(minutes=args.min_interval) else None
+
+
 def safe_emit(args, line, detail, out):
     """Report a failed run; if even that fails, print the line alone."""
     try:
@@ -285,8 +321,19 @@ def run(args, clients, out=sys.stdout):
     -------
     int
         Exit code: 0 done, 1 some writes failed, 2 the read or the run failed,
-        3 too many changes, 4 done but some writes are parked.
+        3 too many changes, 4 done but some writes are parked, 5 too soon
+        after the last run that wrote.
     """
+    try:
+        last = too_soon(args)
+    except Exception as exc:  # noqa: BLE001  the summary names the error type only
+        safe_emit(args, f"run failed: {type(exc).__name__}", traceback.format_exc(), out)
+        return 2
+    if last:
+        print(
+            f"skipped: the last run that wrote started at {last}, less than {args.min_interval:g} minutes ago", file=out
+        )
+        return 5
     try:
         snap = github.read_all(clients)
     except Exception as exc:  # noqa: BLE001  the summary names the error type only
@@ -315,10 +362,16 @@ def main(argv=None, env=os.environ):
     parser.add_argument("--limit", type=int, default=400, help="most writes in one run; the rest wait")
     parser.add_argument("--pause", type=float, default=0.0, help="seconds between writes")
     parser.add_argument("--deadline", type=float, default=600.0, help="seconds from start after which writes stop")
+    parser.add_argument(
+        "--min-interval",
+        type=float,
+        default=50.0,
+        help="minutes from the start of the last run that wrote to the next run with --apply",
+    )
     parser.add_argument("--summary", action="store_true", help="print counts only, no issue data")
     parser.add_argument("--report", help="file for the full report with issue data")
     args = parser.parse_args(argv)
-    args.started = time.monotonic()
+    args.started, args.started_at = time.monotonic(), datetime.now(UTC)
     if env.get("GITHUB_ACTIONS") == "true":
         args.summary = True  # a job log can be public
     old = signal.signal(signal.SIGTERM, _terminate)

board_sync/config.py

@@ -76,5 +76,9 @@
 # treated as a new, empty field, not as a mass clear.
 RESET_MIN = 5
 
+# A board place of an issue that held at least this many synced values and
+# now reads none is held back for the run, not read as a clear of each field.
+WIPE_MIN = 2
+
 # An issue that is on no board for this many days is dropped from the state.
 GONE_DAYS = 90

board_sync/core.py

@@ -35,13 +35,16 @@ class Plan:
         Values that cannot be written because the place lacks the option.
     resets : list of tuple
         ``(place_key, fkey)`` pairs treated as new fields in this run.
+    held : list of tuple
+        ``(issue, place)`` pairs held back in this run, see ``held_places``.
     """
 
     writes: list = field(default_factory=list)
     state: dict = field(default_factory=dict)
     conflicts: list = field(default_factory=list)
     skipped: list = field(default_factory=list)
     resets: list = field(default_factory=list)
+    held: list = field(default_factory=list)
 
 
 def to_place(fkey, place, value):
@@ -231,6 +234,50 @@ def reset_places(snap, state):
     return reset
 
 
+def held_places(snap, state, reset):
+    """Find places of an issue whose synced values read empty with no sign of a clear.
+
+    A board counts when its synced fields held at least ``config.WIPE_MIN``
+    values in the state, not counting reset fields and places marked gone,
+    and every synced field there reads empty now. The issue place counts when
+    one such field reads empty without a removal time, since a clear of an
+    issue field always has one. Such a place is treated as if it did not
+    hold the issue in this run, so the empty read clears no other place; when
+    it reads the same in the next run, it receives the agreed values.
+
+    Parameters
+    ----------
+    snap : dict
+        Snapshot of all places.
+    state : dict
+        State of the previous run.
+    reset : set of tuple
+        ``(place_key, fkey)`` pairs from ``reset_places``.
+
+    Returns
+    -------
+    set of tuple
+        ``(issue, place)`` pairs to hold back.
+    """
+    out, old = set(), state.get("items", {})
+    for iid, item in snap["items"].items():
+        for place in item["places"]:
+            held = filled = lost = 0
+            for fkey, spec in config.FIELDS.items():
+                if place not in scope(snap, item, fkey):
+                    continue
+                value, at = raw_value(item, place, spec["places"][place])
+                filled += value is not None
+                rec = old.get(iid, {}).get(fkey, {})
+                counted = (place_key(item, place), fkey) not in reset and place not in rec.get("gone", ())
+                if counted and rec.get("seen", {}).get(place) is not None:
+                    held += 1
+                    lost += value is None and at is None
+            if lost if place == "issue" else held >= config.WIPE_MIN and filled == 0:
+                out.add((iid, place))
+    return out
+
+
 def _newest(places, times):
     """Pick the place with the newest change time; ties go by ``PLACE_ORDER``."""
     return max(places, key=lambda p: (times[p] or "", -config.PLACE_ORDER.index(p)))
@@ -294,10 +341,11 @@ def decide(fkey, places, current, raw, times, record):
     for p in places:
         before = seen.get(p, _NEW)
         if p in gone and before is not _NEW:
-            # A returning place: empty receives, its old value is no change, and a new
-            # value counts only when it is newer than the agreed value (of known time).
+            # A returning place: its old value is no change, and a new value or a clear counts
+            # only when it is newer than the agreed value (of known time); a board clear has
+            # no time, so an empty board place receives.
             newer = record.get("t") is not None and (times[p] or "") > record["t"]
-            if current[p] is not None and current[p] != before and newer:
+            if current[p] != before and newer:
                 changed.append(p)
         elif before is _NEW:
             # A place that joins with a value counts as a change; an empty one receives.
@@ -333,6 +381,13 @@ def plan(snap, state):
     result = Plan(state={"items": {}, "fields": field_ids(snap, state.get("fields"))})
     reset = reset_places(snap, state)
     result.resets = sorted(reset)
+    held = held_places(snap, state, reset)
+    result.held = sorted(held)
+    if held:
+        items = {iid: dict(item) for iid, item in snap["items"].items()}
+        for iid, place in held:
+            items[iid]["places"] = {p: v for p, v in items[iid]["places"].items() if p != place}
+        snap = {**snap, "items": items}
     old = state.get("items", {})
     now = snap.get("read_at")
     for iid, entry in old.items():