From 548489541575dfd34d4dc4f04a0ea0ece80847bd Mon Sep 17 00:00:00 2001 From: Adrian Pop Date: Tue, 11 Aug 2026 21:34:13 +0200 Subject: [PATCH] Claim a library for a branch, not for a configuration Two machines testing master both claimed five of the same libraries and tested them twice. #303 fixed the confighash being different for no reason; these five differ for a real one. They are the libraries whose reference files come from MAP-LIB_ReferenceResults/v4.1.0, which a maintenance job updates from time to time, and the two machines had fetched them either side of such an update. The libraries pinned to v4.0.0 and v3.2.3, which do not move, agreed. So the configuration really did differ, and the claim was keyed by it: PRIMARY KEY (branch, libname, libversion, omcversion, confighash) which is finer than the question a claim answers. Whether this machine may test this library of this branch has nothing to do with which reference files it has; that belongs to the two questions that are keyed by the configuration and stay as they are - whether results for an exact configuration already exist, and whether two runs' rows can collide. The claim is now one row per (branch, libname). The versions are kept as columns, so a run still says who is testing what and since when. An existing job_claim is narrowed on connect, keeping the freshest claim per library so that a run in progress does not lose its claims and let a second machine in. Checked against a database seeded the way production looks: the live claim survives the migration, the key becomes (branch, libname), a second machine with a different confighash is refused and told who holds it, a free library is granted and then refused to the first machine, and after release the other machine may take it. Plus a sqlite3 run through the CI configuration and a PostgreSQL run with two FMI simulators. --- Generated by Claude Code. --- resultsdb.py | 73 ++++++++++++++++++++++++++++++++++++++-------------- 1 file changed, 54 insertions(+), 19 deletions(-) diff --git a/resultsdb.py b/resultsdb.py index 0d498c9..81bc764 100644 --- a/resultsdb.py +++ b/resultsdb.py @@ -50,18 +50,27 @@ } BRANCH_KEY = ["date", "libname", "model"] +# What a claim is about: this machine is testing this library of this branch. +# Not the configuration it is testing it with - the configuration hash covers +# the reference files, and two machines that fetched them either side of an +# upstream commit would then both claim the same library and both test it, +# which is the duplication the claim exists to prevent. The versions are kept +# to say who is testing what, and the run still decides for itself whether it +# already has results for an exact configuration. +JOB_CLAIM_KEY = ("branch", "libname") + JOB_CLAIM = """ CREATE TABLE IF NOT EXISTS job_claim ( branch text NOT NULL, libname text NOT NULL, - libversion text NOT NULL, - omcversion text NOT NULL, - confighash bigint NOT NULL, + libversion text, + omcversion text, + confighash bigint, host text NOT NULL, state text NOT NULL, claimed_at timestamptz NOT NULL DEFAULT now(), heartbeat timestamptz NOT NULL DEFAULT now(), - PRIMARY KEY (branch, libname, libversion, omcversion, confighash) + PRIMARY KEY (branch, libname) ) """ @@ -283,6 +292,7 @@ def __init__(self, url): self.claims = [] self.heartbeatThread = None self.execute(JOB_CLAIM) + self._migrateJobClaim() self.commit() def _connect(self): @@ -335,6 +345,31 @@ def commit(self): self.conn.commit() self.pending = [] + def _migrateJobClaim(self): + """Narrow an older job_claim to (branch, libname). + + It used to be keyed by the configuration as well, so two machines whose + reference files had been updated at different times each got a claim on the + same library. Claims of a run in progress are kept, the freshest per + library, so migrating does not let a second machine in. + """ + key = [c for (c,) in self.execute("""SELECT a.attname FROM pg_index i + JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey) + WHERE i.indrelid = 'job_claim'::regclass AND i.indisprimary + ORDER BY a.attnum""")] + if tuple(key) == JOB_CLAIM_KEY: + return + print("Narrowing job_claim from (%s) to (%s)" % (", ".join(key), ", ".join(JOB_CLAIM_KEY))) + cursor = self.cursor() + cursor.execute(JOB_CLAIM.replace("job_claim", "job_claim_migrated")) + cursor.execute("""INSERT INTO job_claim_migrated + SELECT DISTINCT ON (branch, libname) branch, libname, libversion, omcversion, + confighash, host, state, claimed_at, heartbeat + FROM job_claim ORDER BY branch, libname, heartbeat DESC""") + cursor.execute("DROP TABLE job_claim") + cursor.execute("ALTER TABLE job_claim_migrated RENAME TO job_claim") + self.commit() + def sql(self, sql, hasParams=False): """sqlite spells the placeholder "?" and psycopg2 spells it "%s". @@ -420,16 +455,18 @@ def claim(self, branch, libname, libversion, omcversion, confighash): A machine that dies stops sending its heartbeat, and after STALE_CLAIM_MINUTES its jobs are up for grabs again. """ - key = (branch, libname, libversion, omcversion, confighash) + key = (branch, libname) got = self.execute("""INSERT INTO job_claim (branch, libname, libversion, omcversion, confighash, host, state) VALUES (?,?,?,?,?,?,'running') - ON CONFLICT (branch, libname, libversion, omcversion, confighash) DO UPDATE - SET host = EXCLUDED.host, state = 'running', claimed_at = now(), heartbeat = now() + ON CONFLICT (branch, libname) DO UPDATE + SET host = EXCLUDED.host, state = 'running', claimed_at = now(), heartbeat = now(), + libversion = EXCLUDED.libversion, omcversion = EXCLUDED.omcversion, + confighash = EXCLUDED.confighash WHERE job_claim.state <> 'running' OR job_claim.heartbeat < now() - interval '%d minutes' RETURNING host""" % STALE_CLAIM_MINUTES, - key + (self.host,)).fetchone() + (branch, libname, libversion, omcversion, confighash, self.host)).fetchone() self.commit() if got is None: return False @@ -439,9 +476,8 @@ def claim(self, branch, libname, libversion, omcversion, confighash): def claimedBy(self, branch, libname, libversion, omcversion, confighash): """The machine holding that job, for the message telling the user why we skip.""" - row = self.execute("""SELECT host, claimed_at FROM job_claim WHERE branch=? AND libname=? - AND libversion=? AND omcversion=? AND confighash=?""", - (branch, libname, libversion, omcversion, confighash)).fetchone() + row = self.execute("SELECT host, claimed_at FROM job_claim WHERE branch=? AND libname=?", + (branch, libname)).fetchone() return row or ("unknown", None) def _startHeartbeat(self): @@ -463,16 +499,15 @@ def _heartbeat(self): import psycopg2 with psycopg2.connect(self.url) as conn: with conn.cursor() as cur: - for (branch, libname, libversion, omcversion, confighash) in list(self.claims): - cur.execute("""UPDATE job_claim SET heartbeat = now() WHERE branch=%s AND libname=%s - AND libversion=%s AND omcversion=%s AND confighash=%s AND host=%s""", - (branch, libname, libversion, omcversion, confighash, self.host)) + for (branch, libname) in list(self.claims): + cur.execute("""UPDATE job_claim SET heartbeat = now() + WHERE branch=%s AND libname=%s AND host=%s""", + (branch, libname, self.host)) def release(self): - for (b, libname, libversion, omcversion, confighash) in self.claims: + for (branch, libname) in self.claims: self.execute("""UPDATE job_claim SET state='done', heartbeat=now() - WHERE branch=? AND libname=? AND libversion=? AND omcversion=? AND confighash=? - AND host=?""", - (b, libname, libversion, omcversion, confighash, self.host)) + WHERE branch=? AND libname=? AND host=?""", + (branch, libname, self.host)) self.claims = [] self.commit()