Refuse a blank owner, and assemble the claim where the owner lives

domains/games/apis/one_d4/BUILD.bazel

@@ -68,6 +68,7 @@ filegroup(
         "migrations/V001__initial_schema.sql",
         "migrations/V002__player_titles.sql",
         "migrations/V003__platform_canonical.sql",
+        "migrations/V004__owner_never_blank.sql",
     ],
 )
 
@@ -559,6 +560,7 @@ java_test_suite(
         "src/test/java/com/muchq/games/one_d4/db/MigrationFilesTest.java",
         "src/test/java/com/muchq/games/one_d4/db/MigrationRunnerTest.java",
         "src/test/java/com/muchq/games/one_d4/db/MigrationTest.java",
+        "src/test/java/com/muchq/games/one_d4/db/OwnerNeverBlankConstraintTest.java",
         "src/test/java/com/muchq/games/one_d4/db/PgTestUrlsTest.java",
         "src/test/java/com/muchq/games/one_d4/db/PlatformCanonicalConstraintTest.java",
         "src/test/java/com/muchq/games/one_d4/db/ReanalysisRequestDaoTest.java",

domains/games/apis/one_d4/migrations/V004__owner_never_blank.sql

@@ -0,0 +1,29 @@
+-- owner_id names the claimant every fence keys on: a heartbeat, a progress
+-- write and a terminal write all match on it, and a claim spends an attempt
+-- only when the owner it presents differs from the one stored. A blank is a
+-- claimant nobody's fence matches and one a re-claim under a blank spends
+-- nothing on — so it is refused on both request tables, whoever the writer
+-- is. NULL stays what it is: no owner.
+--
+-- Blanks are nulled on the one execution that adds the constraint, so a
+-- database holding one converges rather than failing the deploy. A nulled
+-- owner with a lapsed lease is exactly what ClaimNext takes next.
+
+DO $$
+DECLARE
+    target text;
+    constraint_name text;
+BEGIN
+    FOREACH target IN ARRAY ARRAY['indexing_requests', 'reanalysis_requests']
+    LOOP
+        constraint_name := target || '_owner_never_blank';
+        IF NOT EXISTS (
+            SELECT 1 FROM pg_constraint
+            WHERE conrelid = target::regclass AND conname = constraint_name
+        ) THEN
+            EXECUTE format('UPDATE %I SET owner_id = NULL WHERE owner_id = ''''', target);
+            EXECUTE format('ALTER TABLE %I ADD CONSTRAINT %I CHECK (owner_id <> '''')',
+                           target, constraint_name);
+        END IF;
+    END LOOP;
+END $$;

domains/games/apis/one_d4/migrations/manifest.txt

@@ -7,3 +7,4 @@
 V001__initial_schema
 V002__player_titles
 V003__platform_canonical
+V004__owner_never_blank

domains/games/apis/one_d4/src/main/java/com/muchq/games/one_d4/docs/DESIGN.md

@@ -165,7 +165,7 @@ com.muchq.indexer/
 | error_message | TEXT         | Populated on FAILED            |
 | games_indexed | INT          | Running count during processing|
 | exclude_bullet | BOOLEAN     | Part of the live-request key    |
-| owner_id      | VARCHAR(128) | Who holds the lease; the fencing token every write is conditioned on. The Java worker claims as a process, the C++ worker mints one per run so several of its runs can be in flight at once |
+| owner_id      | VARCHAR(128) | Who holds the lease; the fencing token every write is conditioned on, NULL or never blank (V004 CHECK). The Java worker claims as a process, the C++ worker mints one per run so several of its runs can be in flight at once |
 | lease_expires_at | TIMESTAMP | Renewed every 75s while the owner is alive. Past this the request is reclaimable. Deliberately survives a terminal write, as the record of when a worker last held the row |
 | skip_cache    | BOOLEAN      | Persisted so a worker on any instance honours what the submitter asked for |
 | attempts      | INT          | Claims so far. Bounds the requeue loop for a request that keeps killing its worker |

domains/games/apis/one_d4/src/test/java/com/muchq/games/one_d4/db/OwnerNeverBlankConstraintTest.java

@@ -0,0 +1,163 @@
+package com.muchq.games.one_d4.db;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.List;
+import java.util.UUID;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * A blank owner is refused by the request tables, whoever the writer is.
+ *
+ * <p>Every fence keys on {@code owner_id}: a heartbeat, a progress write and a terminal write all
+ * match on it, and a claim spends an attempt only when the owner it presents differs from the one
+ * stored. A row claimed under {@code ''} is one no worker's fence matches and one a re-claim under
+ * {@code ''} does — so it never completes and never retires. {@code NULL} stays what it is: no
+ * owner.
+ */
+public class OwnerNeverBlankConstraintTest {
+
+  private static final List<String> REQUEST_TABLES =
+      List.of("indexing_requests", "reanalysis_requests");
+
+  private TestDb testDb;
+
+  @BeforeEach
+  public void setUp() {
+    testDb = TestDb.create("ownerneverblank");
+  }
+
+  @Test
+  public void everyRequestTableRefusesABlankOwner() throws Exception {
+    for (String table : REQUEST_TABLES) {
+      UUID id = insertRequest(table);
+      assertThatThrownBy(() -> setOwner(table, id, ""))
+          .as("%s should refuse a blank owner", table)
+          .isInstanceOf(SQLException.class)
+          .hasMessageContaining("owner_never_blank");
+    }
+  }
+
+  /** The control: no owner and a named owner are both still storable. */
+  @Test
+  public void aNullOrNamedOwnerIsStillAccepted() throws Exception {
+    for (String table : REQUEST_TABLES) {
+      UUID id = insertRequest(table);
+      assertThatCode(() -> setOwner(table, id, "cpp/host/1/9a1f"))
+          .as("%s should accept a named owner", table)
+          .doesNotThrowAnyException();
+      assertThatCode(() -> setOwner(table, id, null))
+          .as("%s should accept no owner", table)
+          .doesNotThrowAnyException();
+    }
+  }
+
+  /** The constraints are named, so {@code Migration.verify()} notices a database missing them. */
+  @Test
+  public void theConstraintsAreVisibleToBootVerification() {
+    assertThat(constraintNames())
+        .containsExactlyInAnyOrder(
+            "indexing_requests_owner_never_blank", "reanalysis_requests_owner_never_blank");
+  }
+
+  /**
+   * The normalise half. A database holding a blank before the constraint converges instead of
+   * failing the deploy: the step nulls blanks only on the execution that adds the constraint.
+   */
+  @Test
+  public void rerunningTheMigrationClearsABlankOwnerStoredBeforeTheConstraint() throws Exception {
+    for (String table : REQUEST_TABLES) {
+      exec("ALTER TABLE " + table + " DROP CONSTRAINT " + table + "_owner_never_blank");
+      UUID id = insertRequest(table);
+      setOwner(table, id, "");
+
+      new Migration(testDb.dataSource()).run();
+
+      assertThat(ownerOf(table, id)).as("%s blank owner after re-run", table).isNull();
+      assertThat(constraintNames()).contains(table + "_owner_never_blank");
+    }
+  }
+
+  /** Re-run safe, like every step here: the file executes on every deploy. */
+  @Test
+  public void theStepIsIdempotent() {
+    new Migration(testDb.dataSource()).run();
+    new Migration(testDb.dataSource()).run();
+
+    assertThat(constraintNames()).hasSize(REQUEST_TABLES.size());
+  }
+
+  /** A pending row on `table`. At most one reanalysis row per test: its live index allows one. */
+  private UUID insertRequest(String table) throws SQLException {
+    UUID id = UUID.randomUUID();
+    String sql =
+        switch (table) {
+          case "indexing_requests" ->
+              "INSERT INTO indexing_requests (id, player, platform, start_month, end_month)"
+                  + " VALUES (?, ?, 'CHESS_COM', '2024-01', '2024-01')";
+          case "reanalysis_requests" -> "INSERT INTO reanalysis_requests (id) VALUES (?)";
+          default -> throw new IllegalArgumentException(table);
+        };
+    try (Connection conn = testDb.dataSource().getConnection();
+        var stmt = conn.prepareStatement(sql)) {
+      stmt.setObject(1, id);
+      if (table.equals("indexing_requests")) {
+        stmt.setString(2, UUID.randomUUID().toString());
+      }
+      stmt.executeUpdate();
+    }
+    return id;
+  }
+
+  private void setOwner(String table, UUID id, String owner) throws SQLException {
+    try (Connection conn = testDb.dataSource().getConnection();
+        var stmt = conn.prepareStatement("UPDATE " + table + " SET owner_id = ? WHERE id = ?")) {
+      stmt.setString(1, owner);
+      stmt.setObject(2, id);
+      stmt.executeUpdate();
+    }
+  }
+
+  private String ownerOf(String table, UUID id) throws SQLException {
+    try (Connection conn = testDb.dataSource().getConnection();
+        var stmt = conn.prepareStatement("SELECT owner_id FROM " + table + " WHERE id = ?")) {
+      stmt.setObject(1, id);
+      try (var rs = stmt.executeQuery()) {
+        assertThat(rs.next()).as("row %s should still exist", id).isTrue();
+        return rs.getString(1);
+      }
+    }
+  }
+
+  private void exec(String sql) throws SQLException {
+    try (Connection conn = testDb.dataSource().getConnection();
+        var stmt = conn.createStatement()) {
+      stmt.execute(sql);
+    }
+  }
+
+  private List<String> constraintNames() {
+    try (Connection conn = testDb.dataSource().getConnection();
+        var stmt =
+            conn.prepareStatement(
+                "SELECT conname FROM pg_constraint c JOIN pg_namespace n ON n.oid ="
+                    + " c.connamespace WHERE n.nspname = ? AND conname LIKE"
+                    + " '%owner_never_blank'")) {
+      stmt.setString(1, testDb.schema());
+      var names = new java.util.ArrayList<String>();
+      try (var rs = stmt.executeQuery()) {
+        while (rs.next()) {
+          names.add(rs.getString(1));
+        }
+      }
+      return names;
+    } catch (SQLException e) {
+      throw new RuntimeException(e);
+    }
+  }
+}

domains/games/apis/one_d4_worker/BUILD.bazel

@@ -281,10 +281,16 @@ cc_test(
     ],
     tags = ["requires-postgres"],
     deps = [
+        ":index_pool",
+        ":metrics",
         ":migration_files",
         ":pg_queue",
         ":pg_test_db",
+        ":poller",
+        "//domains/platform/libs/futility/otel:capturing_metrics_recorder",
         "//domains/platform/libs/pg",
+        "@com_google_absl//absl/status",
+        "@com_google_absl//absl/status:statusor",
         "@com_google_absl//absl/strings",
         "@com_google_absl//absl/time",
         "@googletest//:gtest_main",

domains/games/apis/one_d4_worker/pg_queue.cc

@@ -50,6 +50,12 @@ std::string TextArray(absl::Span<const std::string> values) {
 
 absl::StatusOr<std::optional<IndexJob>> PgQueue::ClaimNext(
     std::string_view owner, absl::Duration lease, absl::Span<const std::string> at_capacity) {
+  // Every write after this one is fenced on the owner, and a re-claim
+  // under the id already on the row spends no attempt. A blank would be
+  // an owner nobody's fence matches and nobody's re-claim spends — so it
+  // is refused here, before the row is touched, and the schema refuses it
+  // too (V004).
+  if (owner.empty()) return absl::InvalidArgumentError("a claim needs an owner to fence on");
   // One conditional UPDATE, so two workers racing for the same row cannot
   // both win: the row lock decides, and the loser's WHERE no longer
   // matches. FOR UPDATE SKIP LOCKED picks the candidate without the two of

domains/games/apis/one_d4_worker/pg_queue_test.cc

@@ -3,15 +3,21 @@
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
+#include <atomic>
 #include <cstdlib>
 #include <string>
 
 #include "absl/status/status.h"
 #include "absl/status/statusor.h"
 #include "absl/strings/str_cat.h"
 #include "absl/strings/str_format.h"
+#include "absl/time/clock.h"
+#include "domains/games/apis/one_d4_worker/index_pool.h"
+#include "domains/games/apis/one_d4_worker/metrics.h"
 #include "domains/games/apis/one_d4_worker/migration_files.h"
 #include "domains/games/apis/one_d4_worker/pg_test_db.h"
+#include "domains/games/apis/one_d4_worker/poller.h"
+#include "domains/platform/libs/futility/otel/capturing_metrics_recorder.h"
 #include "domains/platform/libs/pg/pg.h"
 
 namespace one_d4_worker {
@@ -422,5 +428,66 @@ TEST_F(PgQueueTest, ProgressIsRefusedOnceTheLeaseHasExpired) {
   EXPECT_EQ(Column(Id(1), "games_indexed"), "0");
 }
 
+// The production wiring over the production statements: a pool of owned
+// connections, the platform gate in front of the claim, a heartbeat and a
+// progress report inside the run, and the terminal write after it. Every one
+// of those is fenced on the id the row was claimed under, so this is where a
+// claim made under any other id shows up — as a lease nobody took.
+TEST_F(PgQueueTest, ARunTakenThroughTheGateIsFencedAllTheWayToCompleted) {
+  InsertOn(Id(1), "alireza", "LICHESS");
+
+  std::atomic<bool> kept{false};
+  std::atomic<bool> reported{false};
+  std::atomic<int> runs{0};
+  const Poller::Run run = [&](const Claim&, LeaseKeeper& lease) -> absl::StatusOr<RunReport> {
+    kept = lease.Keep();
+    reported = lease.Report(7);
+    ++runs;
+    RunReport report;
+    report.games_indexed = 7;
+    return report;
+  };
+
+  PlatformAdmission admission({{"LICHESS", 1}});
+  Poller::Options poller;
+  poller.owner = "cpp/test/1";
+  poller.lease = absl::Minutes(5);
+  poller.admission = &admission;
+  futility::otel::CapturingMetricsRecorder recorder;
+  WorkerMetrics metrics(recorder);
+  IndexPool::Options pool_options;
+  pool_options.slots = 2;
+  pool_options.idle_wait = absl::Milliseconds(10);
+  IndexPool pool([this] { return NewOwnedPgQueue(conninfo_, kMaxAttempts); }, run, poller, metrics,
+                 pool_options);
+
+  // Bounded, so a run that never completes fails here rather than as a
+  // timeout on the whole target with the reason buried in the log.
+  const absl::Time deadline = absl::Now() + absl::Seconds(10);
+  pool.Run([&] { return runs.load() >= 1 || absl::Now() > deadline; },
+           [](absl::Duration wait) { absl::SleepFor(wait); });
+
+  ASSERT_EQ(runs.load(), 1) << "no run completed before the deadline";
+  EXPECT_TRUE(kept.load()) << "the heartbeat did not match the row the claim wrote";
+  EXPECT_TRUE(reported.load());
+  EXPECT_EQ(Column(Id(1), "status"), "COMPLETED");
+  EXPECT_EQ(Column(Id(1), "games_indexed"), "7");
+  EXPECT_EQ(Column(Id(1), "owner_id"), "(null)");
+  EXPECT_EQ(Column(Id(1), "attempts"), "1");
+}
+
+// An owner is what every later write is fenced on, so a blank one is refused
+// before the row is touched rather than written and then matched by nobody.
+TEST_F(PgQueueTest, RefusesToClaimUnderABlankOwner) {
+  Insert(Id(1), "hikaru");
+
+  const auto claimed = queue_->ClaimNext("", absl::Minutes(5));
+
+  EXPECT_EQ(claimed.status().code(), absl::StatusCode::kInvalidArgument) << claimed.status();
+  EXPECT_EQ(Column(Id(1), "status"), "PENDING");
+  EXPECT_EQ(Column(Id(1), "owner_id"), "(null)");
+  EXPECT_EQ(Column(Id(1), "attempts"), "0");
+}
+
 }  // namespace
 }  // namespace one_d4_worker

domains/games/apis/one_d4_worker/poller.cc

@@ -121,9 +121,9 @@ std::string_view ToString(RunOutcome outcome) {
 Poller::Poller(IndexQueue& queue, Run run, Options options)
     : queue_(queue), run_(std::move(run)), options_(std::move(options)) {}
 
-absl::StatusOr<std::optional<Claim>> PlatformAdmission::Claim(
-    absl::FunctionRef<absl::StatusOr<std::optional<IndexJob>>(absl::Span<const std::string>)> claim,
-    std::string_view owner) {
+absl::StatusOr<std::optional<Admitted>> PlatformAdmission::Claim(
+    absl::FunctionRef<absl::StatusOr<std::optional<IndexJob>>(absl::Span<const std::string>)>
+        claim) {
   const absl::MutexLock lock(mu_);
   std::vector<std::string> at_capacity;
   for (const auto& [platform, limit] : limits_) {
@@ -143,8 +143,7 @@ absl::StatusOr<std::optional<Claim>> PlatformAdmission::Claim(
   // Owns nothing; the deleter is the whole point. It runs when the last
   // copy of the claim is destroyed, wherever that happens to be.
   PlatformSlot slot(nullptr, [this, platform](void*) { Release(platform); });
-  return one_d4_worker::Claim{
-      .job = **claimed, .owner = std::string(owner), .slot = std::move(slot)};
+  return Admitted{.job = **claimed, .slot = std::move(slot)};
 }
 
 void PlatformAdmission::Release(const std::string& platform) {
@@ -160,12 +159,24 @@ absl::StatusOr<std::optional<Claim>> Poller::ClaimOne() {
   const auto take = [&](absl::Span<const std::string> at_capacity) {
     return queue_.ClaimNext(owner, options_.lease, at_capacity);
   };
-  if (options_.admission != nullptr) return options_.admission->Claim(take, owner);
-
-  const absl::StatusOr<std::optional<IndexJob>> claimed = take({});
-  if (!claimed.ok()) return claimed.status();
-  if (!claimed->has_value()) return std::nullopt;
-  return Claim{.job = **claimed, .owner = std::move(owner), .slot = nullptr};
+  // Either way the queue reads `owner` in place and it is moved once, here,
+  // after the last read — the Claim is assembled by the one function that
+  // holds the string.
+  absl::StatusOr<std::optional<Admitted>> admitted;
+  if (options_.admission != nullptr) {
+    admitted = options_.admission->Claim(take);
+  } else {
+    absl::StatusOr<std::optional<IndexJob>> claimed = take({});
+    if (!claimed.ok()) return claimed.status();
+    admitted = claimed->has_value()
+                   ? std::optional<Admitted>(Admitted{.job = std::move(**claimed), .slot = nullptr})
+                   : std::nullopt;
+  }
+  if (!admitted.ok()) return admitted.status();
+  if (!admitted->has_value()) return std::nullopt;
+  return Claim{.job = std::move((*admitted)->job),
+               .owner = std::move(owner),
+               .slot = std::move((*admitted)->slot)};
 }
 
 absl::StatusOr<RunOutcome> Poller::RunClaimed(const Claim& claim) {