`ProcessMessageRetentionScheduledTask` crashes with `Unknown table` when two workers run retention concurrently (`drop_table` has no `IF EXISTS`)

Please let me know if I am out of line for copy-pasting info from Claude, but we found it very helpful.

## Describe the bug

When more than one worker process is running, the raw-message retention cleanup can run **concurrently against the same server's message database**. Because `Provisioner#drop_table` issues a bare `DROP TABLE` (no `IF EXISTS`), the second worker to reach an already-dropped table crashes with a `Mysql2::Error (Unknown table ...)` instead of treating it as a no-op.

The retention task aborts with an error like:

```
Mysql2::Error (Unknown table 'postal-server-.raw-')
```

In the logs it typically appears right after the `Tidying raw messages (by size) for ` line, following the `(by days)` line for the *same* server/ID. It recurs on the same few servers/databases.

This only affects a small number of high-volume servers and happens intermittently, which matches the conditions required to trigger it (described below).

## Affected version

Observed on `main` at commit `d038eaa` (2026-06-03). The relevant code paths are long-standing, so earlier/later versions are likely affected too.

Reporter is running many worker processes (the multi-worker precondition is satisfied).

## Root cause

**1. The only code path that drops raw tables is this task.**
`remove_raw_table` → `drop_table` is reached only from `remove_raw_tables_older_than` and `remove_raw_tables_until_less_than_size` (plus `provision`/`clean`, which are dev/test only). Message processing never drops raw tables. Therefore a *double* drop of the same table requires **two concurrent executions of the retention task against the same database**.

**2. `drop_table` is not idempotent** (`lib/postal/message_db/provisioner.rb:64`):

```ruby
def drop_table(table_name)
@database.query("DROP TABLE `#{@database.database_name}`.`#{table_name}`")
end
```

A bare `DROP TABLE` on a table that is already gone raises `Unknown table` rather than no-opping.

Note the asymmetry: the **create** path already defends against exactly this concurrency (`create_raw_table`, `lib/postal/message_db/provisioner.rb:83`):

```ruby
rescue Mysql2::Error => e
# Don't worry if the table already exists, another thread has already run this code.
raise unless e.message =~ /already exists/
end
```

So the codebase already anticipates concurrent provisioner execution — `create` was guarded, but `drop` never received the symmetric guard.

**3. How two executions come to overlap.**
The worker runs scheduled tasks on a single "tasks" thread that holds the `:tasks` role via `WorkerRole.acquire`. The lock is renewed **once per `run_tasks` cycle** (at the top), then *all* tasks run sequentially, then the thread sleeps `task_sleep_time` (60s) and loops. The steal threshold is hardcoded (`app/models/worker_role.rb:30`):

```ruby
updates = where(role: role).where("acquired_at is null OR acquired_at < ?", 5.minutes.ago)
.update_all(acquired_at: Time.current, worker: Postal.locker_name)
```

If a full task cycle takes longer than ~5 minutes (e.g. a large server's raw cleanup), the current holder does not renew in time, so another worker **steals** the `:tasks` role mid-run. Because the holder hasn't finished the task yet, `next_run_after` hasn't been bumped, so the stealing worker sees the retention task as still due and **runs it too** — concurrently, against the same server databases.

## Suggested fixes

**1. Immediate, low-risk — make the drop idempotent.**
Either use `DROP TABLE IF EXISTS` in `drop_table`, or rescue the "Unknown table" `Mysql2::Error` in `remove_raw_table`, mirroring the existing guard in `create_raw_table`. This converts the crash into a harmless no-op and is in the spirit of the existing code. Example:

```ruby
def drop_table(table_name)
@database.query("DROP TABLE IF EXISTS `#{@database.database_name}`.`#{table_name}`")
end
```

or:

```ruby
def remove_raw_table(table)
@database.query("UPDATE ... WHERE raw_table = '#{table}'")
@database.query("DELETE FROM ... WHERE table_name = '#{table}'")
drop_table(table)
rescue Mysql2::Error => e
# Another worker already removed this table concurrently.
raise unless e.message =~ /Unknown table/
end
```

**2. Root cause — prevent two workers from running the same database's retention at once.**
Renew the `:tasks` lock *during* the task loop (a heartbeat / renew inside `TASKS.each`) instead of only between cycles, so a long-running cycle can't be stolen. Alternatively, take a per-task or per-database lock around the retention work. Renewing mid-run is the smaller change and directly closes the steal window. (Simply raising the `5.minutes` threshold is a blunt mitigation — it also delays legitimate takeover when a worker genuinely dies.)

**3. Secondary latent bug worth fixing in the same area.**
`remove_raw_tables_until_less_than_size` loops `until @database.total_size <= size` with `table = tables.shift`. `total_size` sums the `raw_message_sizes` bookkeeping table, not actual table sizes. If that sum never drops to the target (e.g. orphaned `raw_message_sizes` rows whose tables no longer exist) or the real tables are exhausted, `shift` eventually returns `nil`, leading to `remove_raw_table(nil)` → `drop_table(nil)` and another query error. Guarding against an empty list / `nil` table (and/or deriving size from actual tables) would make the loop robust.

```ruby
def remove_raw_tables_until_less_than_size(size)
tables = raw_tables(nil)
tables_removed = []
until @database.total_size <= size
table = tables.shift
break if table.nil? # nothing left to remove; avoid dropping a nil table
tables_removed << table
remove_raw_table(table)
end
tables_removed
end
```

Start with drop_table in lib/postal/message_db/provisioner.rb and trace its callers from ProcessMessageRetentionScheduledTask, then read the task-lock behavior in app/models/worker_role.rb. Reproduce or inspect concurrent retention against the same database, and confirm that an already-removed raw table no longer aborts the task while the existing cleanup behavior remains intact.

Written by the indexing model from the issue text.