Fencing writes downstream

A bulkhead lease can run out while the first callback is still working. Let the store you write to refuse the late write with a fence number.

Every code block comes from examples/12-fencing-downstream.php (opens in a new tab). The example needs the pdo_sqlite extension.

The problem

A bulkhead slot has a lease. If a worker stalls for longer than the lease, the slot becomes free and another worker takes the job. The first worker is not stopped. When it wakes up, it still holds its callback and may write its result. Two workers then believe they own the job. See Bulkhead and cooldown.

Reqtide checks before each dispatch that the lease still outlasts the attempt. That check cannot cover a stall after the dispatch. Reqtide also has no fencing token to give you. A token only helps if the system that receives the write checks it, so the check belongs there.

The idea

Keep a number next to the data you protect. Call it the fence. It only goes up, and only the store changes it.

A worker claims the job. The store adds one to the fence and returns the new value.
The worker does its work and sends the result together with the fence it got.
The store accepts the write only if the fence in the request is still the newest one.

A worker that lost its slot holds an old fence. Its write finds a newer number and is refused.

Claim and write

The two functions below are the whole guard. The claim and the check live in the same store as the data, so one row decides who is current.

PHP

$store = new \PDO('sqlite::memory:', null, null, [\PDO::ATTR_ERRMODE => \PDO::ERRMODE_EXCEPTION]);
$store->exec('CREATE TABLE jobs (id TEXT PRIMARY KEY, fence INTEGER NOT NULL, result TEXT)');
$store->exec("INSERT INTO jobs (id, fence, result) VALUES ('invoice-7', 0, NULL)");

/** Takes the job and returns its new fence. The number only goes up, and only the store hands it out. */
$claim = static function (string $job) use ($store): int {
    $store->beginTransaction();
    $store->prepare('UPDATE jobs SET fence = fence + 1 WHERE id = ?')->execute([$job]);
    $select = $store->prepare('SELECT fence FROM jobs WHERE id = ?');
    $select->execute([$job]);
    $fence = (int) $select->fetchColumn();
    $store->commit();

    return $fence;
};

/** Writes the result only if the fence is still the newest one. Returns false for a stale holder. */
$write = static function (string $job, int $fence, string $result) use ($store): bool {
    $update = $store->prepare('UPDATE jobs SET result = ? WHERE id = ? AND fence = ?');
    $update->execute([$result, $job, $fence]);

    return $update->rowCount() === 1;
};

Use the same idea with any store that can run a conditional write: a version column in SQL, a compare-and-set in a key-value store, or an If-Match header on an HTTP API.

Use it inside a pipeline

Claim the fence inside the callable, so every attempt gets its own. Pass it to every write the attempt makes.

PHP

$pipeline = PipelineBuilder::profile('invoices', 'v1')
    ->partition('shop', 'prod', 'invoice-writer')
    ->attemptTimeout(Duration::seconds(2), AttemptTimeoutRequirement::Cooperative)
    ->bulkhead(1, Duration::seconds(10))
    ->compileInMemory(ExceptionClassifier::create());

The example plays both workers. Worker A claims the job and stalls. Worker B runs through the pipeline and writes.

PHP

// Worker A takes the job and gets fence 1. Then it stalls: a slow call, a paused process, a long garbage collection.
$staleFence = $claim('invoice-7');

echo "-- 1. The current holder writes\n";
$outcome = $pipeline->execute(
    static function (AttemptContext $attempt) use ($claim, $write): string {
        $fence = $claim('invoice-7');

        return $write('invoice-7', $fence, 'written by ' . $attempt->operationId->value) ? "accepted with fence {$fence}" : "refused with fence {$fence}";
    },
    $pipeline->newContext(Duration::seconds(30), operationId: 'worker-b'),
);
echo $outcome, "\n";

Worker A wakes up and tries to write with fence 1.

PHP

// Worker A wakes up. Its lease ran out long ago, and worker B has taken the job with a newer fence.
echo $write('invoice-7', $staleFence, 'written by worker-a') ? "accepted with fence {$staleFence}" : "refused with fence {$staleFence}", "\n";

The output:

Text

-- 1. The current holder writes
accepted with fence 2
-- 2. A holder that lost its slot writes late
refused with fence 1
stored: fence 2, written by worker-b

What this does not cover

The fence protects the data in the store that hands it out. A side effect somewhere else, such as an email or a call to a third party, is not covered. Give those calls an idempotency key.
A write that does not carry the fence is not checked. Send the fence with every write of the attempt.
A worker that was replaced may still use time and connections until it notices. The fence only keeps its result out of the data.