Drain consumer queues at shutdown and make the lookup pause claim atomic

[tricrotism]

Summary

Rows queued shortly before a server stop can be lost: the consumer makes exactly one more pass over one of its two buffers, and that pass does nothing if a lookup holds the consumer pause. Separately, two lookups can both claim the pause, and the first to finish releases it while the other is still reading. This change drains both buffers until they are empty (bounded at 30 seconds) and makes the pause claim atomic.

The problem

Shutdown drops queued rows

Consumer.run (consumer/Consumer.java:485) loops while serverRunning. When the server stops it sets lastRun and goes round once more, then exits. That last pass:

• only processes the current buffer. The consumer alternates two buffers (consumer_id 0 and 1), so rows sitting in the other one, from a pass that deferred or from producers that were still queueing when the buffer switched, are never written;
• returns immediately at Consumer.java:467 if isPaused is set, which is the case whenever a lookup, rollback or API lookup is running at stop time. The final pass then writes nothing at all;
• on SQLite can wait 500 ms in Database.getConnection (database/Database.java:420) for that same pause to clear, then get null and write nothing;
• after an exception sleeps 30 seconds in errorDelay() before the thread exits, holding up the stop.

Scenario: a staff member runs /co lookup or a plugin polls the API while the server restarts. Every block, container and chat row queued in the last seconds before the stop is gone.

Lookups can overlap

Every lookup claims the pause like this (database/Lookup.java:62-65, database/LookupRaw.java:88-91 and 169-173):

while (Consumer.isPaused && !Consumer.isPersistenceHalted()) {
    Thread.sleep(1);
}
Consumer.isPaused = true;

Two lookup threads can both see false and both set true. When the first one finishes it sets isPaused = false, and the consumer starts a write pass while the second lookup is still reading. On SQLite that is the lock contention the pause exists to prevent.

The fix

consumer/Consumer.java

• After shutdown starts, the loop keeps going while either buffer has rows or outstanding reservations (consumer_id[i][1]), up to a 30-second deadline.
• A held lookup pause is respected until the last 5 seconds of that window. After that the pass runs anyway, because losing the rows is worse than overlapping a lookup that is still running while the server stops.
• If persistence is halted, the loop exits instead of spinning, and the last passes skip errorDelay().
• New claimLookupPause(): waits for the flag to clear and sets it inside one monitor, so two lookups cannot both claim it.

consumer/process/Process.java, database/Database.java

• The consumer fetches its connection through a new Database.getConsumerConnection(waitTime, ignorePause). On the shutdown passes it skips the SQLite wait on isPaused, since Consumer.run has already decided whether the pause blocks the pass. Every other caller of getConnection is unchanged.

database/Lookup.java, database/LookupRaw.java

• The four check-then-set sequences call Consumer.claimLookupPause().

Behaviour change

• Server stop can take up to 30 seconds longer when rows are still queued. With an empty queue it is as fast as before.
• Lookups that start at the same moment now run one after the other. They already waited for each other when the timing was not tight, so this only closes the race.

Risk

• A forced pass in the last 5 seconds runs while a lookup still holds the pause. Process sets and clears isPaused around its own pass, so that lookup's claim can be cleared early. This only happens during the last 5 seconds of a stop, and the alternative is losing the queued rows.
• claimLookupPause polls with wait(1), the same 1 ms granularity as the old Thread.sleep(1) loop. Nothing notifies it, so there is no missed-wakeup case to reason about.
• The drain reads consumer_id state through the existing synchronized map, and only after serverRunning is false.

Testing

Build: mvn package passes.
Row parity: the 47-step scenario, which ends by stopping the server, on Paper 26.2 and Folia 1.21.11 with SQLite. Folia matched upstream row for row. Paper differed only by the random plant that bone meal grows. No new errors. The scenario does not hold a lookup open across the stop, so it shows no regression rather than the lost-rows case.

Suggested test: on SQLite, have a plugin queue a few thousand block changes and start a long /co lookup in the same tick, then /stop. Count the rows in the database after restart. Upstream loses the queued rows. This branch writes them, and the stop finishes within the 30-second window.

[netlify]

❌ Deploy Preview for coreprotect failed. Why did it fail? →

Name	Link
🔨 Latest commit	998b7bd
🔍 Latest deploy log	https://app.netlify.com/projects/coreprotect/deploys/6ab3ee54bed6750008810898

[Intelli]

Thanks -- automated review is requesting the following changes:

• Preserve pause ownership between lookups and the consumer. Forcing a write after 25 seconds lets either operation clear the other’s pause. The consumer also needs to check/claim the pause after acquiring the lifecycle lock, rather than receiving an earlier boolean snapshot.
• Make shutdown waits respect the intended deadline, or describe the timeout accurately as a limit between passes. Reservation waits, lock acquisition, and database operations can currently exceed it.
• Integrate the drain outcome with ShutdownService. If the deadline expires with records still queued, report incomplete persistence and the remaining queue size instead of silently ending the consumer and proceeding through normal shutdown completion.

Also, the description should say six lookup pause-acquisition sites are updated, rather than four.

[Intelli]

The following changes are requested:

• Include outstanding reservations in the incomplete-drain decision. A reservation-only timeout currently reports zero unsaved items and produces the normal shutdown-success message. Reservations can be reported separately from published queue entries.
• Make the DuckDB lock-upgrade cleanup release only a successfully acquired lock. An interrupted timed write-lock acquisition currently reaches unlock() without ownership and throws IllegalMonitorStateException.