144 lines
6.7 KiB
Markdown
144 lines
6.7 KiB
Markdown
|
# T0.5 — EventLog port + redb implementation
|
|||
|
|
|
|||
|
|
| Field | Value |
|
|||
|
|
|---|---|
|
|||
|
|
| Phase | P0 — Foundations |
|
|||
|
|
| Size | L — over 3 days |
|
|||
|
|
| Status | Not started |
|
|||
|
|
| Flags | — |
|
|||
|
|
| Spec | inlined below |
|
|||
|
|
| Blocks | T0.6, T1.x |
|
|||
|
|
|
|||
|
|
## Goal
|
|||
|
|
|
|||
|
|
The full `EventLog` port and its embedded `redb` implementation. Keys
|
|||
|
|
`(TenantId, RunId, BranchId, Lsn)`. Async signatures even though the local
|
|||
|
|
implementation is synchronous underneath.
|
|||
|
|
|
|||
|
|
## Facts (inlined — no spec read needed)
|
|||
|
|
|
|||
|
|
```rust
|
|||
|
|
#[async_trait]
|
|||
|
|
pub trait EventLog: Send + Sync {
|
|||
|
|
/// Append records, apply derived state, advance consumer position, enqueue
|
|||
|
|
/// the outbox — all four or none. Returns the assigned LSNs.
|
|||
|
|
async fn commit(&self, batch: CommitBatch) -> Result<Vec<Lsn>>;
|
|||
|
|
|
|||
|
|
async fn read(&self, key: BranchKey, from: Lsn, limit: usize) -> Result<Vec<LogRecord>>;
|
|||
|
|
|
|||
|
|
async fn put_checkpoint(&self, key: BranchKey, upto: Lsn, state: &[u8]) -> Result<()>;
|
|||
|
|
/// Newest checkpoint at or below `upto`. `None` means fold from LSN 0.
|
|||
|
|
async fn latest_checkpoint(&self, key: BranchKey, upto: Lsn)
|
|||
|
|
-> Result<Option<Checkpoint>>;
|
|||
|
|
|
|||
|
|
/// Committed-but-unshipped export intents, in `(BranchKey, Lsn)` order.
|
|||
|
|
/// Read by the relay, never from the execution path.
|
|||
|
|
async fn drain_outbox(&self, tenant: TenantId, limit: usize) -> Result<Vec<OutboxEntry>>;
|
|||
|
|
async fn ack_outbox(&self, shipped: &[(BranchKey, Lsn)]) -> Result<()>;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
pub struct CommitBatch {
|
|||
|
|
/// Carries the tenant. Never a separate parameter beside a key that already
|
|||
|
|
/// holds one — two sources for one fact is one too many.
|
|||
|
|
pub key: BranchKey,
|
|||
|
|
pub records: Vec<WorkEvent>, // LSNs assigned by the implementation
|
|||
|
|
pub state: Vec<StateDelta>,
|
|||
|
|
pub position: Option<ConsumerPosition>,
|
|||
|
|
pub outbox: Vec<ExportIntent>,
|
|||
|
|
}
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
- **`commit` is one method, not four, because atomicity is the contract.** A port
|
|||
|
|
exposing `append` alone puts the other three writes outside the transaction —
|
|||
|
|
the durability guarantee gone, and gone invisibly, since each write
|
|||
|
|
individually succeeds.
|
|||
|
|
- Everything is `async`. An async signature over a local call costs a negligible
|
|||
|
|
poll; a sync signature over a network call is impossible, and the Postgres
|
|||
|
|
backend (T7.1) implements this same trait. The prior revision declared a
|
|||
|
|
synchronous store and documented in the same file that a networked
|
|||
|
|
implementation could not honour the signature — a port finished while already
|
|||
|
|
known to be unimplementable.
|
|||
|
|
- `redb` specifics: pure Rust, ACID, MVCC, copy-on-write shadow paging, so a torn
|
|||
|
|
write cannot corrupt the file — it simply does not take effect. The durability
|
|||
|
|
enum is `#[non_exhaustive]`, so set `Durability::Immediate` **explicitly**
|
|||
|
|
rather than relying on the default. (Avoid `sled` — years at 0.34 beta with
|
|||
|
|
known space amplification.)
|
|||
|
|
- LSNs are per branch, monotonic, gap-free. Not a global counter: a global one
|
|||
|
|
serializes every run through a single atomic, and the ordering contract only
|
|||
|
|
promises per-run total order.
|
|||
|
|
- `latest_checkpoint` and `drain_outbox`/`ack_outbox` exist from the start. A
|
|||
|
|
checkpoint that can be written and not read is an optimization nobody can use;
|
|||
|
|
writes without their matching reads are how T2.5 and T7.4 discover the port is
|
|||
|
|
short.
|
|||
|
|
|
|||
|
|
## Steps
|
|||
|
|
|
|||
|
|
1. Define the port trait, `CommitBatch`, `Checkpoint`, `OutboxEntry`,
|
|||
|
|
`ExportIntent`, `StateDelta`, `ConsumerPosition` exactly as above.
|
|||
|
|
2. Open the four `redb` tables: `EVENT_LOG`, `RUN_STATE`, `CONSUMER_POSITION`,
|
|||
|
|
`OUTBOX`. Every key is `BranchKey`-derived or `Scoped<_>` (T0.1).
|
|||
|
|
3. Implement LSN allocation **inside** the write transaction: read the current
|
|||
|
|
max LSN for the `BranchKey`, assign `max+1..` across the batch. Being inside
|
|||
|
|
the transaction is what makes it gap-free under concurrency.
|
|||
|
|
4. Implement `commit` as a single `begin_write` → four inserts → `commit()`, with
|
|||
|
|
`Durability::Immediate` set explicitly.
|
|||
|
|
5. Implement `read` as a range scan bounded by `(key, from)..`, honouring `limit`.
|
|||
|
|
6. Implement the checkpoint pair and the outbox pair. `drain_outbox` returns in
|
|||
|
|
`(BranchKey, Lsn)` order — that ordering is the only one the relay is promised.
|
|||
|
|
7. Write the conformance suite as a **generic function over `impl EventLog`**, so
|
|||
|
|
T7.1's Postgres backend runs the identical tests. This is where the
|
|||
|
|
abstraction becomes real or does not.
|
|||
|
|
|
|||
|
|
## Acceptance
|
|||
|
|
|
|||
|
|
- Append 10k records across 3 branches of 2 runs in 2 tenants; read back in order
|
|||
|
|
per branch; assert no cross-tenant visibility.
|
|||
|
|
- Port audit: compile stub call sites for T2.5 (checkpoints), T7.4 (outbox relay)
|
|||
|
|
and T8.4 (reduction) against the trait. Every method they need already exists.
|
|||
|
|
|
|||
|
|
## Verify
|
|||
|
|
|
|||
|
|
**Harness:** the conformance suite written as
|
|||
|
|
`fn conformance<L: EventLog>(make: impl Fn() -> L)`. It is the deliverable T7.1
|
|||
|
|
reuses unmodified — write it as a library function, not as `#[test]` bodies.
|
|||
|
|
|
|||
|
|
**Integration test** — `tests/it_eventlog_conformance.rs`:
|
|||
|
|
1. Append 10k records across 3 branches × 2 runs × 2 tenants.
|
|||
|
|
2. Read back per branch from LSN 0; assert contiguous `1..n` with no gaps and no
|
|||
|
|
duplicates.
|
|||
|
|
3. **Concurrency:** spawn 16 tasks committing to the *same* `BranchKey`
|
|||
|
|
simultaneously; assert the union of returned LSNs is exactly `1..=total` with
|
|||
|
|
no repeats. Gap-freedom only breaks under contention.
|
|||
|
|
4. **Isolation:** scan the raw table with no tenant predicate and assert every
|
|||
|
|
key carries the expected tenant — do not rely on the filtered read path.
|
|||
|
|
5. Round-trip a checkpoint: `put_checkpoint`, then `latest_checkpoint` at an LSN
|
|||
|
|
above and below it.
|
|||
|
|
6. Outbox: commit with entries, `drain_outbox`, assert `(BranchKey, Lsn)`
|
|||
|
|
ascending order, `ack_outbox`, assert a second drain returns empty.
|
|||
|
|
7. **Port audit:** a `tests/port_audit.rs` with stub call sites for T2.5, T7.4
|
|||
|
|
and T8.4 that must compile against the trait.
|
|||
|
|
|
|||
|
|
**Command:** `cargo test -p storage --test it_eventlog_conformance --test port_audit`
|
|||
|
|
|
|||
|
|
**False pass:**
|
|||
|
|
- Single-threaded LSN allocation test. It passes against a
|
|||
|
|
read-then-write-outside-the-transaction implementation, which is the actual bug.
|
|||
|
|
- Cross-tenant isolation checked through a query that already filters by tenant —
|
|||
|
|
it passes against a completely unscoped table. Step 4 is the real check.
|
|||
|
|
- A conformance suite written inline as `#[test]`s, which then cannot be reused
|
|||
|
|
for Postgres and quietly becomes redb-only.
|
|||
|
|
|
|||
|
|
## Traps
|
|||
|
|
|
|||
|
|
- Adding a convenience `append()` that skips the other three tables. It will be
|
|||
|
|
on the hot path within a week.
|
|||
|
|
- Passing `tenant` as a second parameter beside `BranchKey`.
|
|||
|
|
- Letting `Durability` default. The enum is `#[non_exhaustive]`; the default can
|
|||
|
|
change underneath you.
|
|||
|
|
|
|||
|
|
---
|
|||
|
|
|
|||
|
|
Background (not required to do this task):
|
|||
|
|
[rust-agentic-sys.md](../../../rust-agentic-sys.md) §7, §8.3 ·
|
|||
|
|
[rust-agentic-task.md](../../../rust-agentic-task.md)
|