The Lock Held; the Count Counted Intent, Not Writes

The short version

Even with FOR UPDATE serialising, counting a summary by the length of the classified intent can leave the rows correctly at one while the summary totals two. The TV liveness checker (runTvLivenessCheck) carried a flaky test where two concurrent runs made newlyDown total 2 where it should be 1. On the DB side, unresolved rows were always kept at one by the parent row’s FOR UPDATE lock, and the row-count assertion passed stably. What was broken was how the summary counted: it returned the number of TVs classified as down targets rather than the number actually INSERTed, so the run that skipped its INSERT because of the lock was counted as well.

Where it started

The checker splits into classifyTvLiveness, a pure function deciding each TV’s liveness, and applyTransitions, which reflects that decision in the DB. The caller, runTvLivenessCheck, returned the length of the decision as the summary.

// equivalent to the pre-fix code
const classification = classifyTvLiveness(states, now, thresholds);
await applyTransitions(tx, classification);

return {
  scanned: states.length,
  newlyDown: classification.newlyDown.length,
  recovered: classification.recovered.length,
};

classification.newlyDown is “the list of TVs judged as needing to transition to down”, not the number actually written to the DB. In a test running two connections concurrently, the assertion r1.newlyDown + r2.newlyDown === 1 failed intermittently.

Why

The classification by classifyTvLiveness is based on a snapshot read earlier by loadDeviceStates. When two checker runs fire concurrently, both read the same state — “this TV has no unresolved downtime row yet” — and both classify the same TV as newlyDown.

The actual DB write is serialised inside applyTransitions. Before the down-transition INSERT, the parent tv_devices row is locked FOR UPDATE.

// the FOR UPDATE serialisation point (inside applyTransitions)
await tx
  .select({ deviceId: tvDevices.deviceId })
  .from(tvDevices)
  .where(eq(tvDevices.deviceId, down.deviceId))
  .for("update");

const open = await tx
  .select({ id: tvDeviceDowntime.id })
  .from(tvDeviceDowntime)
  .where(
    and(eq(tvDeviceDowntime.deviceId, down.deviceId), isNull(tvDeviceDowntime.recoveredAt)),
  );
if (open.length > 0) {
  // another checker run INSERTed first → just align the state flag, do not double count.
  await tx
    .update(tvDevices)
    .set({ alertState: "down", updatedAt: new Date() })
    .where(eq(tvDevices.deviceId, down.deviceId));
  continue;
}

Because a parent TV row for an unresolved downtime row always exists via the FK, the lock target is never empty even on a first down (with zero unresolved rows), so the second run waits for the lock, rescans, sees an unresolved row already exists and skips the INSERT. As far as DB consistency goes, this is correct.

The problem is that this skipped run’s down was still in the classification.newlyDown array. Counting the summary as classification.newlyDown.length counts both the winner and the loser as one each — a total of two. Only one row was written to the DB, and only the number the checker returned diverged from reality.

Fixing it

We changed the summary’s count from the length of the classification to what applyTransitions actually wrote.

// after the fix (applyTransitions)
async function applyTransitions(
  tx: TenantTx,
  classification: TvLivenessClassification,
): Promise<{ newlyDown: number; recovered: number }> {
  let newlyDown = 0;
  let recovered = 0;
  for (const down of classification.newlyDown) {
    // ...FOR UPDATE lock, recheck for unresolved rows...
    if (open.length > 0) {
      // ...align the state flag and continue (not counted)
      continue;
    }
    await tx.insert(tvDeviceDowntime).values({ /* ... */ });
    // ...
    newlyDown += 1; // count only when the INSERT ran (the skip path already continued)
  }

  for (const rec of classification.recovered) {
    const closed = await tx
      .update(tvDeviceDowntime)
      .set({ /* recoveredAt, durationSec etc. */ })
      .where(and(eq(tvDeviceDowntime.deviceId, rec.deviceId), isNull(tvDeviceDowntime.recoveredAt)))
      .returning({ id: tvDeviceDowntime.id });
    // ...
    if (closed.length > 0) {
      recovered += 1; // count only when an unresolved row was actually closed
    }
  }

  return { newlyDown, recovered };
}

The down side increments newlyDown only when the INSERT actually ran, and not on the path skipped by the lock (continue). The recover side was fixed symmetrically. Since the UPDATE’s condition is “unresolved rows only”, if another checker run closed the same row first the real updated count is zero. Using .returning() there, recovered is incremented only when at least one row comes back. The caller, runTvLivenessCheck, now simply returns those actual write counts as the summary.

// after the fix (runTvLivenessCheck)
return {
  scanned: states.length,
  ...(await applyTransitions(tx, classification)),
};

The winner is 1, the loser 0, and the total is deterministically 1 regardless of timing. No schema or migration change was needed; the existing FOR UPDATE serialisation stands and only the counting of the return value was fixed.

In summary

Whether the lock works and whether the summary’s numbers are right were problems in different layers. The lock here did its job — keeping the DB to one row — correctly from the start; what was broken was separate aggregation code counting “how many were processed” by the length of a decision. When returning a summary or a count from work involving concurrency, you have to write with the distinction always in mind: count what actually happened (writes and updates performed), not what was intended (classifications, intent).