Internals

What this page covers

The on-disk and in-memory state of a Forum, the locking model, and the recovery path on cold start.

State

pub struct Forum {
    strand: Arc<RwLock<Strand>>,
    live_tuples: Arc<RwLock<HashMap<EntryId, Vec<u8>>>>,
}

pub type EntryId = u64;

Field	Role
`strand`	Canonical, append-only write log of `TupleEntry` blocks
`live_tuples`	In-memory map of `EntryId → bytes` for tuples that have not yet been taken

TupleEntry encoding

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum TupleEntry {
    Write { id: EntryId, tuple: Vec<u8> },
    Take { write_id: EntryId },
}

Encoded with serde_json::to_vec. The Strand stores the JSON-encoded bytes; the schema is forward-compatible.

Write path

write(tuple)
   │
   ├─► strand.write().await    (lock 1)
   │   entry_id = strand.len()
   │   serialize TupleEntry::Write { id: entry_id, tuple: tuple.clone() }
   │   strand.append(bytes)
   │   release lock 1
   │
   ├─► live_tuples.write().await  (lock 2)
   │   live_tuples.insert(entry_id, tuple)
   │   release lock 2
   │
   └─► return Ok(entry_id)

The two locks are sequential. There is a brief window between the Strand append and the in-memory insert where a concurrent read would not yet see the new tuple. The Strand append is the linearization point: as soon as it returns, the tuple is durable.

Take path

take(pattern)
   │
   ├─► live_tuples.write().await   (held for the entire op)
   │
   ├─► linear scan: find first (id, tuple) where tuple == pattern.0
   │
   ├─► if found:
   │     live_tuples.remove(id)
   │     strand.write().await
   │     serialize TupleEntry::Take { write_id: id }
   │     strand.append(bytes)
   │     return Some(tuple)
   │
   └─► else: return None

The live_tuples write lock is held across the Strand append. Holding both locks means take serializes with all other Forum operations — concurrent writes, reads, and takes all block until the current take finishes.

For high-write workloads, this is a coarse lock. The current implementation prioritizes correctness; a future revision will replace the linear scan with a content-addressed index and shrink the critical section.

Read path

read(pattern)
   │
   ├─► live_tuples.read().await
   │
   ├─► linear scan: find first (_, tuple) where tuple == pattern.0
   │
   ├─► return Some(tuple.clone()) on match, None otherwise

Read takes only the in-memory read lock. It is O(N) in the size of live_tuples.

Cold-start rebuild

The current public API constructs a Forum with an empty live_tuples map. To populate the index after a restart, walk the Strand from sequence 0 and replay:

use forum::{Forum, TupleEntry};
use strand::Strand;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;

async fn rebuild_live_tuples(
    strand: &Arc<RwLock<Strand>>,
) -> Result<HashMap<u64, Vec<u8>>, Box<dyn std::error::Error>> {
    let strand = strand.read().await;
    let len = strand.len();
    let mut live: HashMap<u64, Vec<u8>> = HashMap::new();

    for i in 0..len {
        let bytes = strand.get(i).await?;
        if let Ok(entry) = serde_json::from_slice::<TupleEntry>(&bytes) {
            match entry {
                TupleEntry::Write { id, tuple } => {
                    live.insert(id, tuple);
                }
                TupleEntry::Take { write_id } => {
                    live.remove(&write_id);
                }
            }
        }
    }

    Ok(live)
}

The weave-sdk performs an equivalent replay on open_forum. If you construct Forum directly, plan to do the same.

Concurrency summary

Operation	Strand lock	live_tuples lock	Held simultaneously?
`write`	write (brief)	write (after strand)	No
`read`	none	read	No
`take`	write (inside critical section)	write (entire op)	Yes

Failure injection points

Drop the Strand append in take between the live_tuples.remove and the strand.append. On the next restart the rebuild will see the tuple as live again. Verify your worker is idempotent.
Concurrent writes during a long-running take. The Strand append will queue; latency rises but correctness is preserved.

Where to extend

Replace the linear scan in read/take with a content-addressed index (e.g. BLAKE3 prefix tree).
Add a blocking take_blocking that uses a tokio::sync::Notify fired on every write.
Add a query method that returns an async stream of matches without consuming.

On this page