Refuse a stale worker's fold, and stop workers going silent while they price groups

erd_queue.py

@@ -3552,17 +3552,134 @@ def finalize_bundle_stats(self, branch_key):
         return (row["n_bundles"], row["max_bundle_nodes"],
                 row["total_bundle_wall_millis"], row["censored_units"])
 
-    def complete_candidate(self, branch_key, idx):
-        """Mark a candidate claim authoritatively complete (done=1)."""
+    def claim_is_current(self, branch_key, idx, claimed_by=None,
+                         bundle_id=None, budget=None):
+        """Does this caller still hold an unfinished claim on this branch?
+
+        One question for the whole set of writes an evaluation produces.  A
+        result carries branch state in several places -- the taint flag, the
+        running best, the cut flag, the nodes spent, and the completion itself
+        -- and every one of them is only meaningful for the branch incarnation
+        the candidate was evaluated against.  Guarding them one at a time
+        cannot be made safe: refusing one while accepting the others leaves the
+        branch describing a mixture of two incarnations, which is how a stale
+        OVER_ERD_LIMIT sets cut_occurred on a replacement that has no ceiling,
+        and finalize then reaches add_cut_result with a NULL bound.
+
+        Answered in one indexed read: the claim must still exist unfinished and
+        belong to this caller, and the branch must still be open at the budget
+        the caller evaluated at.  A NULL stored budget predates the column and
+        is admitted, as everywhere else.
+        """
+        branch_id = self._intern_branch(branch_key)
+        if branch_id is None:
+            return False
+        row = self._conn.execute("""
+            SELECT 1
+            FROM candidate_claims c
+            JOIN active_branches a ON a.branch_id = c.branch_id
+            WHERE c.branch_id = ? AND c.idx = ? AND c.done = 0
+              AND (? IS NULL OR c.claimed_by = ?)
+              AND (? IS NULL OR c.bundle_id = ?)
+              AND a.status = 'open'
+              AND (? IS NULL OR a.budget IS NULL OR a.budget = ?)
+            LIMIT 1
+        """, (branch_id, idx, claimed_by, claimed_by, bundle_id, bundle_id,
+              budget, budget)).fetchone()
+        return row is not None
+
+    def apply_candidate_result(self, branch_key, idx, *, claimed_by=None,
+                               bundle_id=None, budget=None, nodes_spent=0,
+                               infeasible=False, tainted=False, best=None,
+                               cut=False):
+        """Apply every write one candidate evaluation produces, or none.
+
+        Returns True when the result was applied.
+
+        The writes are the branch's nodes spent, its taint flag, its running
+        best, its cut flag, and the candidate's completion.  All five describe
+        the branch incarnation the candidate was evaluated against, and a
+        reclaimed claim can be reissued -- or the branch finalized, deleted and
+        re-created at another budget -- while the evaluation runs.
+
+        Checking first and writing after cannot close that: the check and each
+        write are separate statements, so the branch can change between them
+        and leave a mixture of two incarnations behind.  That is how a stale
+        OVER_ERD_LIMIT sets cut_occurred on a replacement with no ceiling, and
+        finalize then reaches add_cut_result with a NULL bound against a NOT
+        NULL column.  So the validation and the writes share one transaction,
+        and the claim is re-read inside it.
+
+        `best` is (best_guess, best_erd, max_depth) or None.  Each write keeps
+        its own guard as well: they cost nothing here and they still hold for
+        the callers that use them directly.
+        """
+        opened_transaction = not self._conn.in_transaction
+        if opened_transaction:
+            self._conn.execute("BEGIN IMMEDIATE")
+        try:
+            if not self.claim_is_current(branch_key, idx,
+                                         claimed_by=claimed_by,
+                                         bundle_id=bundle_id, budget=budget):
+                applied = False
+            else:
+                if nodes_spent or infeasible:
+                    self.add_nodes_spent(branch_key, nodes_spent,
+                                         infeasible=infeasible)
+                if tainted:
+                    self.mark_branch_tainted(branch_key)
+                if best is not None:
+                    best_guess, best_erd, max_remaining_depth = best
+                    self.update_branch_best(branch_key, best_guess, best_erd,
+                                            max_remaining_depth, budget=budget)
+                if cut:
+                    self.mark_branch_cut(branch_key)
+                self.complete_candidate(branch_key, idx,
+                                        claimed_by=claimed_by,
+                                        bundle_id=bundle_id)
+                applied = True
+        except Exception:
+            if opened_transaction:
+                self._conn.execute("ROLLBACK")
+            raise
+        if opened_transaction:
+            self._conn.execute("COMMIT")
+        return applied
+
+    def complete_candidate(self, branch_key, idx, claimed_by=None,
+                           bundle_id=None):
+        """Mark a candidate claim authoritatively complete (done=1).
+
+        Returns True when the row completed was the caller's own claim.
+
+        Scoped to that claim, because branch_key and idx alone do not identify
+        one.  A worker whose claim was reclaimed while it was still evaluating
+        goes on to finish; by then the index may have been reissued, to another
+        worker on this branch or to a different incarnation of it after a
+        finalize and re-creation.  Completing by key and index alone marks that
+        live claim done while contributing nothing to it, and the branch can
+        then finalize without the candidate ever having been evaluated at the
+        budget it now holds -- cacheing an optimum some candidate beats, or a
+        loss that is not one.
+
+        claimed_by and bundle_id are the pair `claim_next_bundle` stamps: the
+        bundle id is unique to one claim call and settles the case where the
+        same worker re-claimed the same index, and claimed_by carries a bare
+        claim that has no bundle.  Passing neither asks for no check.
+        """
         now = int(time.time())
         branch_id = self._intern_branch(branch_key, create=True)
         self._conn.execute("""
             UPDATE candidate_claims SET done = 1, done_at = ?
             WHERE branch_id = ? AND idx = ?
-        """, (now, branch_id, idx))
+              AND (? IS NULL OR claimed_by = ?)
+              AND (? IS NULL OR bundle_id = ?)
+        """, (now, branch_id, idx, claimed_by, claimed_by,
+              bundle_id, bundle_id))
         n = self._conn.execute("SELECT changes()").fetchone()[0]
         self._tally_wal_traffic(
             'candidate_claims/complete', n, n * _CLAIM_ROW_WAL_BYTES)
+        return n > 0
 
     def complete_bundle_two_level_erd_prunes(self, branch_key, bundle_id,
                                              candidate_indices, nodes_spent=0,
@@ -3650,13 +3767,27 @@ def complete_bundle_two_level_erd_prunes(self, branch_key, bundle_id,
                 updated_branch_count * _CLAIM_ROW_WAL_BYTES)
         return completed_candidate_count
 
-    def update_branch_best(self, branch_key, best_guess, best_erd, max_depth=None):
+    def update_branch_best(self, branch_key, best_guess, best_erd,
+                           max_depth=None, budget=None):
         """Lower the branch's running best (monotone — never raises it).
 
         max_depth is the winning candidate's worst-case line length; it is
         stored atomically with the best it belongs to, so best_max_depth always
         describes the current best_guess.
 
+        budget is the budget the caller evaluated at, and the update applies
+        only to a branch still open at that budget.  A branch can finalize and
+        be re-created at another budget under the same branch_key — the same
+        answer set reached by a second spine of a different length — while a
+        worker holding a claim on the old branch is still evaluating.  Its cost
+        belongs to the budget it was computed at, and a cost from a larger
+        budget is below what a smaller one can achieve, so the monotone test
+        below would accept it and drive the new branch's best under its own
+        optimum.  Ownership and priority both survive the re-creation and so
+        catch nothing.  A stored budget of NULL predates the column and is
+        admitted, matching how callers derive a budget from the spine for
+        those; a caller passing no budget asks for no check.
+
         The same statement stamps first_best_at/nodes_at_first_best on the
         update that creates the branch's first incumbent, and leaves them alone
         on every later improvement — COALESCE keeps the first value, so "how
@@ -3675,8 +3806,9 @@ def update_branch_best(self, branch_key, best_guess, best_erd, max_depth=None):
                 nodes_at_first_best = COALESCE(nodes_at_first_best, nodes_spent)
             WHERE branch_id = ?
               AND (best_erd IS NULL OR ? < best_erd)
+              AND (? IS NULL OR budget IS NULL OR budget = ?)
         """, (best_erd, best_guess, max_depth, now, now, branch_id,
-              best_erd))
+              best_erd, budget, budget))
 
     def read_branch_best(self, branch_key):
         """Return (best_guess, best_erd, ceiling) or (None, None, None).

erd_swarm.py

@@ -549,7 +549,8 @@ def check(self, token, candidate_list, last_index,
         # ceiling, if any, rides on the branch's ceiling column instead).
         if best_guess is not None:
             self._worker.queue.update_branch_best(
-                branch_key, best_guess, best_erd, best_max_remaining_depth)
+                branch_key, best_guess, best_erd, best_max_remaining_depth,
+                budget=budget)
 
         result = self._worker.cooperative_solve(
             branch_words, budget,
@@ -1258,6 +1259,22 @@ def _heartbeat(self, branch_key, n_words, claim_idx, claim_started_at,
         # Count every invocation (one per node) BEFORE the throttle, so the
         # node counter is exact even though we only write every HB_SECONDS.
         self._nodes += 1
+        self._liveness_tick(branch_key, n_words, claim_idx, claim_started_at,
+                            best_guess, best_erd, force=force,
+                            bound_erd=bound_erd)
+
+    def _liveness_tick(self, branch_key, n_words, claim_idx, claim_started_at,
+                       best_guess, best_erd, force=False,
+                       bound_erd=None):
+        """Prove the worker is alive without counting a node.
+
+        `_nodes` means candidate evaluations — the cost model, add_nodes_spent
+        and the accuracy rows all read it as one — so a signal that fires per
+        response group, or anywhere else below a candidate, must come through
+        here instead of `_heartbeat`.  A worker that has not reached this
+        within HB_TIMEOUT_SECONDS has its in-flight claims reclaimed and
+        handed to another worker.
+        """
         now = time.time()
         if not force and now - self._last_hb < HB_SECONDS:
             return
@@ -1601,6 +1618,9 @@ def _metric_observer(group_sizes, has_self, candidate_cost_lower_bound,
             branch_floor_table=self.branch_floor_table,
             hint_cache=self.hint_cache,
             heartbeat=lambda: self._heartbeat(
+                branch_key, n_words, idx, claim_started,
+                local_candidate, local_best, bound_erd=_eff_bound()),
+            liveness_tick=lambda: self._liveness_tick(
                 branch_key, n_words, idx, claim_started,
                 local_candidate, local_best, bound_erd=_eff_bound()))
         cand_elapsed = time.time() - cand_t0
@@ -1611,15 +1631,11 @@ def _metric_observer(group_sizes, has_self, candidate_cost_lower_bound,
                            idx, cand_elapsed, status, self._cand_max_depth)
 
         nodes_delta = self._nodes - nodes_before
-        if self._adaptive and (nodes_delta > 0 or status == OVER_DEPTH_BUDGET):
-            # These counters cover candidates proven infeasible at this level,
-            # not candidates whose taint arrived from a deeper branch.  They
-            # are therefore a lower bound on local infeasibility.  Every such
-            # proof also carries budget_tainted, so infeasible_candidates > 0
-            # implies that the branch is marked tainted below.
-            self.queue.add_nodes_spent(
-                branch_key, nodes_delta,
-                infeasible=status == OVER_DEPTH_BUDGET)
+        # Counted with the rest of the result rather than on its own.  An
+        # aborted candidate keeps its claim open for another worker to redo, so
+        # charging its nodes here as well would count the same candidate twice.
+        record_nodes = (self._adaptive
+                        and (nodes_delta > 0 or status == OVER_DEPTH_BUDGET))
 
         candidate_outcome = {
             SOLVED: 'exact',
@@ -1655,32 +1671,48 @@ def _record_candidate_accuracy():
             _record_candidate_accuracy()
             return False
 
-        # A candidate excluded by the depth cap (anywhere in its subtree)
-        # taints the branch: its ERD is only valid at this budget.  Marked
-        # for any candidate, winner or not — see the taint rule.
-        if budget_tainted:
-            self.queue.mark_branch_tainted(branch_key)
+        # What this evaluation has to say about the branch, decided before any
+        # of it is written.  A candidate excluded by the depth cap (anywhere in
+        # its subtree) taints the branch: its ERD is only valid at this budget,
+        # and that holds for any candidate, winner or not — see the taint rule.
+        improved_best = None
+        mark_cut = False
         if status == SOLVED:
             self.n_ok += 1
             if local_best is None or cost < local_best:
                 local_best, local_candidate, local_md = cost, candidate, cand_md
-                self.queue.update_branch_best(branch_key, local_candidate,
-                                              local_best, local_md)
+                improved_best = (local_candidate, local_best, local_md)
                 shared_best = local_best
         elif status == OVER_ERD_LIMIT:
             self.n_cutoff += 1
-            if branch_ceiling is not None:
-                # Priced out on a ceilinged branch.  Only consulted at finalize
-                # when best_guess is NULL — where no real best ever existed, so
-                # every price-out was against the ceiling and the branch is a
-                # cut, not a proven loss.
-                self.queue.mark_branch_cut(branch_key)
+            # Priced out on a ceilinged branch.  Only consulted at finalize
+            # when best_guess is NULL — where no real best ever existed, so
+            # every price-out was against the ceiling and the branch is a cut,
+            # not a proven loss.
+            mark_cut = branch_ceiling is not None
         elif status == OVER_DEPTH_BUDGET:
             self.n_pruned += 1
         else:  # pragma: no cover
             self.n_useless += 1
 
-        self.queue.complete_candidate(branch_key, idx)
+        # One transaction: the claim is re-read inside it, so either every one
+        # of these lands on the incarnation this candidate was evaluated
+        # against or none of them lands at all.
+        if not self.queue.apply_candidate_result(
+                branch_key, idx, claimed_by=self.name, bundle_id=bundle_id,
+                budget=budget,
+                nodes_spent=nodes_delta if record_nodes else 0,
+                infeasible=record_nodes and status == OVER_DEPTH_BUDGET,
+                tainted=budget_tainted, best=improved_best, cut=mark_cut):
+            # The claim was reissued, or the branch was re-created, while this
+            # candidate ran.  The bundle is not abandoned with it: a one-level
+            # prune sweep replaces a single claim row, so the siblings may
+            # still be this worker's to finish, and a worker that is alive and
+            # heartbeating never has them reclaimed for it.
+            logger.warning(
+                '%s lost candidate %s (idx=%d) mid-evaluation; its result '
+                'describes a branch incarnation this worker no longer holds '
+                'and was discarded', self.name, candidate, idx)
         # The outbound claim telemetry is required for branch ETA reporting,
         # regardless of whether this worker uses adaptive decomposition.
         now_complete = time.time()
@@ -1794,7 +1826,10 @@ def _complete_bundle_two_level_erd_prunes(
                 words, candidate, self.rcache, guesses=self.all_words,
                 pattern_matrix=self.pattern_matrix,
                 branch_indices=branch_indices,
-                branch_floor_table=self.branch_floor_table)
+                branch_floor_table=self.branch_floor_table,
+                liveness_tick=lambda: self._liveness_tick(
+                    branch_key, n_words, candidate_index, claim_started_at,
+                    best_guess, best_erd, bound_erd=bound_erd))
             if candidate_cost_lower_bound >= bound_erd:
                 pruned_candidate_indices.append(candidate_index)