Take Tuple

What this page covers

The take operation: what it does, why it is the right primitive for work queues, and how to reason about concurrent takers.

take(&Pattern) finds at most one live tuple matching the pattern, removes it from the in-memory index, appends a TupleEntry::Take tombstone to the Strand, and returns the consumed bytes. If no matching tuple is live, the return is Ok(None).

use std::sync::Arc;
use tokio::sync::RwLock;
use strand::{Strand, StrandConfig};
use forum::{Forum, Pattern};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let dir = tempfile::tempdir()?;
    let cfg = StrandConfig::default().with_storage(dir.path());
    let strand = Arc::new(RwLock::new(Strand::new(cfg).await?));
    let forum = Forum::new(strand);

    forum.write(b"job-1".to_vec()).await?;
    forum.write(b"job-1".to_vec()).await?; // two entries with identical payload

    // Take consumes exactly one.
    let first = forum.take(&Pattern(b"job-1".to_vec())).await?;
    assert_eq!(first.as_deref(), Some(b"job-1".as_slice()));

    // The second take consumes the other entry.
    let second = forum.take(&Pattern(b"job-1".to_vec())).await?;
    assert_eq!(second.as_deref(), Some(b"job-1".as_slice()));

    // No more matches.
    let third = forum.take(&Pattern(b"job-1".to_vec())).await?;
    assert!(third.is_none());

    Ok(())
}

Atomicity

take holds the live_tuples write lock for the full duration of the match-and-remove. Two concurrent take calls for the same pattern see different tuples (or one sees None); they never see the same tuple twice.

The Strand append for the Take tombstone happens after the in-memory removal but inside the same critical section. If the Strand append fails, the removal is not rolled back — the tuple is gone from the in-memory index even though the durable tombstone is missing. This is the current behavior; treat take failures as a signal that the local index may diverge from the Strand and rebuild from the log on the next process start.

Concurrent consumers

Consumers	Behavior
Multiple `take` callers on one process	Serialized by the `live_tuples` write lock; one of them wins
Multiple processes (different machines) sharing a strand via replication	Each process maintains its own `live_tuples` index; both may take the same tuple locally before the tombstone replicates. Use this carefully for cross-process work queues.

Warning

Across processes, two peers can both take the same tuple before the Take tombstone replicates. Each process maintains its own live_tuples index, so both will return Some with the same payload. Worker handlers must be idempotent, or use the coordinator-claim pattern from Distributed Coordination.

For cross-process queues, the standard mitigation is to design tuples with a claim_token field and have the workflow re-verify the claim against a Lens-stored authority before performing irreversible work.

Errors

Variant	Cause	Recovery
`ForumError::Strand`	Strand append failed when writing the tombstone	Resume work but note that the in-memory and Strand views may have diverged; rebuild on restart
`ForumError::Serialization`	`TupleEntry::Take` could not be encoded	Should not happen for valid `EntryId`s

Patterns

Worker loop

use forum::{Forum, Pattern};

async fn run_worker(forum: &Forum, kind_prefix: &[u8]) -> Result<(), Box<dyn std::error::Error>> {
    loop {
        match forum.take(&Pattern(kind_prefix.to_vec())).await? {
            Some(tuple) => handle(&tuple).await?,
            None => {
                tokio::time::sleep(std::time::Duration::from_millis(50)).await;
            }
        }
    }
}

async fn handle(_tuple: &[u8]) -> Result<(), Box<dyn std::error::Error>> {
    // your job logic here
    Ok(())
}

The poll loop is required because take does not block today. A blocking variant is on the roadmap; until then, exponential backoff is the right default.

Safe consume with retry

async fn take_with_retry(
    forum: &Forum,
    pattern: &Pattern,
    max_attempts: usize,
) -> Result<Option<Vec<u8>>, Box<dyn std::error::Error>> {
    for attempt in 0..max_attempts {
        match forum.take(pattern).await {
            Ok(result) => return Ok(result),
            Err(e) if attempt + 1 < max_attempts => {
                tracing::warn!(?e, attempt, "take failed, retrying");
                tokio::time::sleep(std::time::Duration::from_millis(100 * (1 << attempt))).await;
            }
            Err(e) => return Err(e.into()),
        }
    }
    Ok(None)
}