🐛 fix(async-rw): attribute read-write holds to their owners - #838

The read-write locks attributed holds to whichever thread ran the call. A release or re-entry from the wrong place therefore acted on a hold that belonged to someone else. SoftReadWriteLock also kept granting levels on a hold its heartbeat (the thread that refreshes the holder record) had already reported lost. 🔒 This PR tracks soft read levels per thread, records the async wrappers' sync-lock hold under an owner no thread can have, and refuses to add levels to a lost hold. It builds on #833.

A thread with no read level could end another thread's soft read hold

SoftReadWriteLock shared one _Hold between every reader thread, with a plain level counter. Any thread's release() decremented it. At zero it committed the process out of the generation log (the on-disk list of who holds the lock). The cross-thread check that #761 added covers write holds only, while ReadWriteLock got per-thread read depths in #806. A balanced with rw.read_lock(): reached the stray release whenever something dropped its hold under it. The multi-threading how-to allows two such drops: release(force=True) from another thread, and an on_compromise callback that releases. A second thread then took a fresh hold. The first thread's block exit took it away, and a writer in another process acquired the lock while the second thread was still reading, with compromise still None.

sequenceDiagram
    participant A as thread 1
    participant B as thread 2
    participant L as SoftReadWriteLock
    participant P as another process
    A->>L: read_lock() enter (hold X, level 1)
    L->>L: release(force=True) from elsewhere: hold X gone
    B->>L: read_lock() enter (hold Y, level 1)
    rect rgb(255, 214, 214)
        Note over A,L: before
        A->>L: with-exit release(): level 0, leaves the log
        P->>L: acquire_write(): granted beside thread 2
    end
    rect rgb(216, 243, 220)
        Note over A,L: after
        A->>L: with-exit release(): RuntimeError, hold Y untouched
        P->>L: acquire_write(blocking=False): Timeout
    end

_Hold now keeps read_depths per thread. The check in _raise_if_not_holder() refuses a non-forced release from a thread that holds no level of the current hold, matching the twin class. The hold's own heartbeat thread stays exempt, since on_compromise runs there. That thread owns no level, so its release gives back one of a reader's.

Sync code on the executor worker re-entered the tasks' hold

Both async wrappers took their sync-lock level by running the sync lock's public acquire on an executor thread. The sync lock recorded that pool worker as the owner. With executor=None, AsyncSoftReadWriteLock uses the loop's default executor, the same pool that asyncio.to_thread() uses, and a ThreadPoolExecutor hands new work to an idle worker. Sync code that landed on the worker which had taken the level passed the thread check. SoftReadWriteLock(path).write_lock() entered beside a task's write_lock(), and a sync release() dropped the tasks' hold, with no exception on either side. AsyncReadWriteLock had the same gap through a caller-supplied executor or a recycled thread id. The per-task table from #746 fixed this between tasks, not between a task and sync code.

flowchart LR
    T[task acquire_write on worker W] --> O1[owner = W's thread id]
    S[sync write_lock via to_thread lands on W] --> C{owner check}
    O1 --> C
    C -- "same thread (before)" --> IN[second writer inside]
    O2[owner = task id 0] --> C2{owner check}
    S2[sync write_lock on W] --> C2
    C2 -- "W is not 0 (after)" --> R[RuntimeError]
    classDef ok fill:#d8f3dc,stroke:#2d6a4f,color:#1b4332;
    classDef bad fill:#ffd6d6,stroke:#c1121f,color:#5c0000;
    classDef info fill:#cfe4ff,stroke:#1f6feb,color:#0b1f3a;
    class T,S,S2,C,C2 info;
    class O1,IN bad;
    class O2,R ok;

The wrappers now enter through _acquire_level_of(), which records the level under _TASKS_THREAD_ID, the value 0 that no thread identifier takes, in both sync classes. The sync lock then refuses a sync write acquire and any unforced release on the worker while tasks hold, and refuses a task's write acquire while a thread holds; a read acquire from either side joins the other's read level, as from any thread. The thread recorded at acquire time is no longer needed, so backend_thread is gone.

The soft wrapper's release dropped sync readers of a shared singleton

Since #761 AsyncSoftReadWriteLock left its sync lock with release(force=True) as its cross-thread release. On the default singleton, a synchronous SoftReadWriteLock(path).read_lock() in the same process nests into the same _Hold. The forced release set its level to 0 and committed the process out of the snapshot. The sync reader lost its lock when the last task left, or when the caller cancelled an in-flight task acquire. AsyncReadWriteLock had the same bug fixed in #815; this class did not receive it.

sequenceDiagram
    participant S as sync reader thread
    participant T as asyncio task
    participant L as SoftReadWriteLock (singleton)
    participant P as another process
    S->>L: read_lock() enter (level 1)
    T->>L: acquire_read() on worker (level 2)
    rect rgb(255, 214, 214)
        Note over T,L: before: release(force=True) sets level 0, leaves the log
        T->>L: release()
        P->>L: acquire_write(): granted
    end
    rect rgb(216, 243, 220)
        Note over T,L: after: give back the tasks' one level
        T->>L: release()
        P->>L: acquire_write(blocking=False): Timeout
    end

Both the release path and the cancelled-acquire rollback now leave through _leave_backend(), which gives back the tasks' one level through _release_level_of(_TASKS_THREAD_ID), the way AsyncReadWriteLock already did. The per-thread read depths from the first section make a one-level release attributable.

A lost soft hold kept admitting new readers

Once the heartbeat found the participant gone from the latest snapshot, _report_compromise() stored the loss and ran on_compromise. It left the hold in place until the last level went, a shape from #738. While the hold stood, the lock treated every acquire as a re-entry. Any thread of the process, and any new task of AsyncSoftReadWriteLock through the table, got the read lock without consulting the generation log while a writer in another process held the lock. With overlapping reader threads the level stayed above zero, so the state persisted.

stateDiagram-v2
    direction LR
    classDef ok fill:#d8f3dc,stroke:#2d6a4f,color:#1b4332
    classDef bad fill:#ffd6d6,stroke:#c1121f,color:#5c0000
    classDef info fill:#cfe4ff,stroke:#1f6feb,color:#0b1f3a
    Held: hold stands
    Lost: compromise recorded, hold still set
    Free: no hold
    Held --> Lost: heartbeat sees eviction
    Lost --> Lost: acquire joins (before) / raises (after)
    Lost --> Free: last level released
    Free --> Held: acquire via the log
    class Held ok
    class Lost bad
    class Free info

The re-entry check in _validate_reentrant() now raises RuntimeError once the heartbeat has recorded a loss, under the same internal lock the report writes under. Publishing a new hold resets the loss, so a loss seen there belongs to this hold. The async wrapper raises the same error from its admission check, since the table answers a new reader without reaching the sync lock. Raising is the smallest change. Detaching the lost hold instead would need the old holders' pending releases not to land on the new one.

Behavior changes

SoftReadWriteLock.release() from a thread holding no level of a shared read lock raises RuntimeError, as the how-to now states; release(force=True) and close() still work from any thread. Acquiring on a soft hold with a recorded loss raises RuntimeError until the holder releases it. A sync caller refused because tasks hold the write lock sees held by thread 0 in the message.