sources = pendingPartSources.get(part);
+ if (sources == null || sources.isEmpty()) {
+ throw new IllegalStateException("Reassembled part has no lifecycle source ownership");
+ }
+ return sources;
}
@Override
diff --git a/sendium-core/src/main/java/gr/cytech/sendium/core/storage/OutboundStage.java b/sendium-core/src/main/java/gr/cytech/sendium/core/storage/OutboundStage.java
new file mode 100644
index 00000000..7d01d9fb
--- /dev/null
+++ b/sendium-core/src/main/java/gr/cytech/sendium/core/storage/OutboundStage.java
@@ -0,0 +1,42 @@
+package gr.cytech.sendium.core.storage;
+
+import java.util.Objects;
+
+/**
+ * An explicitly owned, single-use storage lifecycle. Construction does not start a stage.
+ * Operations require READY; closing is idempotent and does not imply terminal completion.
+ * Implementations must support concurrent operation calls while open.
+ *
+ * Worker/filter mutations belong to execution state and are preserved through explicit transitions
+ * and returns. They must not silently change an earlier-stage record through a shared mutable object.
+ * Recorded payload and processing stage must stay consistent. Snapshot/mapping/ownership mechanisms
+ * are implementation choices, not a requirement to copy on every operation. Custom message subtypes
+ * and their fields must be preserved.
+ */
+public interface OutboundStage extends AutoCloseable {
+ void open();
+
+ Status status();
+
+ @Override
+ void close();
+
+ enum Role {
+ PENDING, ROUTER_QUEUE, ROUTED_WORK
+ }
+
+ enum State {
+ NEW, READY, STOPPING, CLOSED, FAILED
+ }
+
+ /** Backend names are implementation identities, not the list of supported standalone profiles. */
+ record Status(String backend, boolean durable, State state) {
+ public Status {
+ Objects.requireNonNull(backend, "backend");
+ Objects.requireNonNull(state, "state");
+ if (backend.isBlank()) {
+ throw new IllegalArgumentException("Backend identity must not be blank");
+ }
+ }
+ }
+}
diff --git a/sendium-core/src/main/java/gr/cytech/sendium/core/storage/OutboundStorageException.java b/sendium-core/src/main/java/gr/cytech/sendium/core/storage/OutboundStorageException.java
new file mode 100644
index 00000000..c18c803b
--- /dev/null
+++ b/sendium-core/src/main/java/gr/cytech/sendium/core/storage/OutboundStorageException.java
@@ -0,0 +1,31 @@
+package gr.cytech.sendium.core.storage;
+
+import java.util.Objects;
+
+/** Operation failures must describe the boundary without including message content or credentials. */
+public class OutboundStorageException extends RuntimeException {
+ private final OutboundStage.Role stage;
+ private final Reason reason;
+
+ public OutboundStorageException(OutboundStage.Role stage, Reason reason, String message) {
+ this(stage, reason, message, null);
+ }
+
+ public OutboundStorageException(OutboundStage.Role stage, Reason reason, String message, Throwable cause) {
+ super(Objects.requireNonNull(message, "message"), cause);
+ this.stage = Objects.requireNonNull(stage, "stage");
+ this.reason = Objects.requireNonNull(reason, "reason");
+ }
+
+ public OutboundStage.Role stage() {
+ return stage;
+ }
+
+ public Reason reason() {
+ return reason;
+ }
+
+ public enum Reason {
+ UNAVAILABLE, CAPACITY_EXCEEDED, OWNERSHIP_CONFLICT, UNSUPPORTED, INVALID_TRANSITION
+ }
+}
diff --git a/sendium-core/src/main/java/gr/cytech/sendium/core/storage/PendingMessageStore.java b/sendium-core/src/main/java/gr/cytech/sendium/core/storage/PendingMessageStore.java
new file mode 100644
index 00000000..dd15e829
--- /dev/null
+++ b/sendium-core/src/main/java/gr/cytech/sendium/core/storage/PendingMessageStore.java
@@ -0,0 +1,37 @@
+package gr.cytech.sendium.core.storage;
+
+import gr.cytech.sendium.core.message.StandardMessage;
+import gr.cytech.sendium.core.outbound.OutboundWork.SourceId;
+
+import java.util.Set;
+
+public interface PendingMessageStore extends OutboundStage {
+ /**
+ * Records accepted source state before protocol success. Later execution mutations must not
+ * silently change that state. Repeating the same source ID while it remains pending is idempotent
+ * and never overwrites its accepted content. A failure must
+ * not be acknowledged. This admits ready work, subject to the selection eligibility policy.
+ * Repeating either admission method preserves the first accepted payload and held/ready disposition.
+ * Deduplication after terminal source removal is not promised.
+ */
+ void admit(SourceId source, M message);
+
+ /** Retains an accepted source without making it eligible for selection until makeHeldReady succeeds. */
+ void admitHeld(SourceId source, M message);
+
+ /**
+ * Makes held sources ready using one execution message referring to their nonempty source set. Original
+ * source records remain authoritative. Publication must bind the whole set before exposing work
+ * to selection; it does not bypass selection limits or create another accepted source.
+ * While all sources remain retained, repeating the exact source set preserves the first published
+ * payload and cannot duplicate work, including after selection/routing. Missing, ordinarily admitted,
+ * or differently grouped sources fail explicitly. The caller owns assembly/expiry policy.
+ */
+ void makeHeldReady(Set sources, M message);
+
+ /**
+ * Called only after all work and required handoffs are terminal. Missing sources are already
+ * complete. Partial multi-source completion is allowed; retrying the full set must be safe.
+ */
+ void complete(Set sources);
+}
diff --git a/sendium-core/src/main/java/gr/cytech/sendium/core/storage/RoutedWorkStore.java b/sendium-core/src/main/java/gr/cytech/sendium/core/storage/RoutedWorkStore.java
new file mode 100644
index 00000000..2b2a374d
--- /dev/null
+++ b/sendium-core/src/main/java/gr/cytech/sendium/core/storage/RoutedWorkStore.java
@@ -0,0 +1,55 @@
+package gr.cytech.sendium.core.storage;
+
+import gr.cytech.sendium.core.message.StandardMessage;
+import gr.cytech.sendium.core.outbound.OutboundWork.Assignment;
+import gr.cytech.sendium.core.outbound.OutboundWork.Completed;
+import gr.cytech.sendium.core.outbound.OutboundWork.Routed;
+import gr.cytech.sendium.core.outbound.OutboundWork.SelectionId;
+import gr.cytech.sendium.core.outbound.OutboundWork.WorkId;
+
+import java.time.Duration;
+import java.util.Optional;
+
+public interface RoutedWorkStore extends OutboundStage {
+ /**
+ * Records the execution payload and one destination consistently before its work may be taken.
+ * Future recovery resumes this recorded work at its worker, not through routing again.
+ * While retained, repeating a selection returns its original assignment and work ID without reactivating
+ * completed work or overwriting content. Callers retry with the same assignment;
+ * a selection has exactly one recorded destination and work identity.
+ * Conflicting source identities or destinations for an existing selection fail as INVALID_TRANSITION.
+ * After forget, the coordinator must reject stale recording attempts using its lifecycle ownership.
+ */
+ Routed record(Assignment assignment);
+
+ /**
+ * Takes one work projection without completing it or silently changing earlier-stage records.
+ * Zero timeout polls; negative timeouts are invalid. Destination must be nonblank.
+ * A work item has at most one active taker.
+ * Retain the returned Routed value as the local attempt handle when returning unfinished work.
+ */
+ Optional> take(String destination, Duration timeout) throws InterruptedException;
+
+ /**
+ * Retains the recorded destination and updated execution projection when retrying or draining
+ * in-flight work. Identity, selection, and destination must match the take. Repeated release cannot
+ * duplicate work or reactivate completed work. Earlier-stage state cannot silently change;
+ * preserving runtime mutations does not promise durable retry counters or timing.
+ * Pass the take result itself with its message updated, not a reconstructed projection.
+ */
+ void release(Routed work);
+
+ /**
+ * Called only after all provider parts and required handoffs for the active work are terminal.
+ * Provider-part aggregation belongs to the provider-processing boundary, not separate routed items.
+ * Returns the completed selection for source cleanup. Until forget, repeating completion of the terminal work
+ * returns the same result so cleanup can be retried without sending again. Unknown IDs fail explicitly.
+ */
+ Completed complete(WorkId work);
+
+ /**
+ * Discards terminal bookkeeping only after pending and selected completion succeed. Nonterminal
+ * selections cannot be forgotten; repeating forget for an absent selection is harmless.
+ */
+ void forget(SelectionId selection);
+}
diff --git a/sendium-core/src/main/java/gr/cytech/sendium/core/storage/SelectedRouterStore.java b/sendium-core/src/main/java/gr/cytech/sendium/core/storage/SelectedRouterStore.java
new file mode 100644
index 00000000..e0185099
--- /dev/null
+++ b/sendium-core/src/main/java/gr/cytech/sendium/core/storage/SelectedRouterStore.java
@@ -0,0 +1,52 @@
+package gr.cytech.sendium.core.storage;
+
+import gr.cytech.sendium.core.message.StandardMessage;
+import gr.cytech.sendium.core.outbound.OutboundWork.Selected;
+import gr.cytech.sendium.core.outbound.OutboundWork.SelectionId;
+
+import java.time.Duration;
+import java.util.Optional;
+
+/**
+ * The implementation receives its pending source and selection/preparation policy through composition.
+ * Selection is not constrained to scanning Sendium-owned records or exposing SQL through this API.
+ */
+public interface SelectedRouterStore extends OutboundStage {
+ /**
+ * Selects at most limit eligible, priority-aware items into the router backlog, returning the newly staged count.
+ * Limit must be positive. Selected, taken, and routed sources are ineligible for another selection.
+ * Source ownership is retained, including after partial publication failure. Stable selection IDs
+ * allow retry/reconciliation without repeating already-committed preparation. No cross-store
+ * transaction is implied. The implementation must not load the entire pending backlog.
+ */
+ int selectToRouter(int limit);
+
+ /**
+ * Takes an execution projection without completing selected or pending ownership.
+ * Execution mutations must not silently modify retained earlier-stage state.
+ * Zero timeout polls; negative timeouts are invalid. An item has at most one active taker.
+ * Retain the returned Selected value as the local attempt handle when returning unfinished work.
+ */
+ Optional> take(Duration timeout) throws InterruptedException;
+
+ /**
+ * Returns taken, not-yet-routed work with its updated execution projection, without silently changing
+ * earlier-stage state. Identity/source ownership must match the take. Repeated release cannot enqueue
+ * duplicates. Preserving runtime mutations does not promise durable retry counters or timing.
+ * Pass the take result itself with its message updated, not a reconstructed or unrelated projection.
+ */
+ void release(Selected selected);
+
+ /**
+ * Stops router scheduling after the single destination has been recorded. This is idempotent and
+ * retains selected state through terminal processing, including for a future durable router
+ * combined with memory-backed destinations. It does not complete the pending source.
+ */
+ void markRouted(SelectionId selection);
+
+ /**
+ * Removes selected ownership after terminal processing and pending completion. Idempotent;
+ * never use this to signal router dequeue or initial destination dispatch.
+ */
+ void complete(SelectionId selection);
+}
diff --git a/sendium-core/src/main/java/gr/cytech/sendium/core/storage/memory/MemoryPendingMessageStore.java b/sendium-core/src/main/java/gr/cytech/sendium/core/storage/memory/MemoryPendingMessageStore.java
new file mode 100644
index 00000000..f9e2abfb
--- /dev/null
+++ b/sendium-core/src/main/java/gr/cytech/sendium/core/storage/memory/MemoryPendingMessageStore.java
@@ -0,0 +1,307 @@
+package gr.cytech.sendium.core.storage.memory;
+
+import gr.cytech.sendium.core.message.StandardMessage;
+import gr.cytech.sendium.core.outbound.OutboundWork.SourceId;
+import gr.cytech.sendium.core.storage.OutboundStorageException;
+import gr.cytech.sendium.core.storage.PendingMessageStore;
+
+import java.time.Instant;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.NavigableSet;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Function;
+import java.util.function.UnaryOperator;
+
+import static gr.cytech.sendium.core.storage.OutboundStorageException.Reason.CAPACITY_EXCEEDED;
+import static gr.cytech.sendium.core.storage.OutboundStorageException.Reason.INVALID_TRANSITION;
+import static gr.cytech.sendium.core.storage.OutboundStorageException.Reason.OWNERSHIP_CONFLICT;
+import static gr.cytech.sendium.core.storage.OutboundStorageException.Reason.UNAVAILABLE;
+import static gr.cytech.sendium.core.storage.OutboundStorageException.Reason.UNSUPPORTED;
+
+/**
+ * Instance-local accepted state with a bounded source count. Snapshot functions must preserve the
+ * concrete message type and detach all mutable fields. Snapshot and eligibility functions run under
+ * the store lock and must be side-effect-free, nonblocking, and must not call back into these stores.
+ */
+public final class MemoryPendingMessageStore implements PendingMessageStore {
+ final ReentrantLock lock = new ReentrantLock();
+ final Condition changed = lock.newCondition();
+
+ private final int capacity;
+ private final UnaryOperator snapshots;
+ private final Function eligibleAt;
+ private final Map> records = new HashMap<>();
+ private final NavigableMap>> waiting = new TreeMap<>(Comparator.reverseOrder());
+ private State state = State.NEW;
+ private Object selectorOwner;
+ private long sequence;
+
+ public MemoryPendingMessageStore(int capacity, UnaryOperator snapshots) {
+ this(capacity, snapshots, message -> Instant.MIN);
+ }
+
+ /** Eligibility is evaluated when ready work is first published; held admission does not evaluate it. */
+ public MemoryPendingMessageStore(int capacity, UnaryOperator snapshots, Function eligibleAt) {
+ if (capacity <= 0) {
+ throw new IllegalArgumentException("Pending capacity must be positive");
+ }
+ this.capacity = capacity;
+ this.snapshots = Objects.requireNonNull(snapshots, "snapshots");
+ this.eligibleAt = Objects.requireNonNull(eligibleAt, "eligibleAt");
+ }
+
+ @Override
+ public void open() {
+ lock.lock();
+ try {
+ if (state == State.CLOSED) {
+ throw new OutboundStorageException(Role.PENDING, INVALID_TRANSITION, "Closed memory pending store cannot reopen");
+ }
+ state = State.READY;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public Status status() {
+ lock.lock();
+ try {
+ return new Status("memory", false, state);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public void admit(SourceId source, M message) {
+ admit(source, message, false);
+ }
+
+ private void admit(SourceId source, M message, boolean held) {
+ Objects.requireNonNull(source, "source");
+ Objects.requireNonNull(message, "message");
+ lock.lock();
+ try {
+ requireReady(Role.PENDING);
+ if (records.containsKey(source)) {
+ return;
+ }
+ if (records.size() >= capacity) {
+ throw new OutboundStorageException(Role.PENDING, CAPACITY_EXCEEDED, "Memory pending capacity reached");
+ }
+ M snapshot = snapshot(message, Role.PENDING);
+ Source accepted = new Source<>(snapshot, held);
+ Entry ready = held ? null : readyEntry(Set.of(source), snapshot);
+ records.put(source, accepted);
+ if (ready != null) {
+ accepted.ready = ready;
+ addWaiting(ready);
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public void admitHeld(SourceId source, M message) {
+ admit(source, message, true);
+ }
+
+ @Override
+ public void makeHeldReady(Set sources, M message) {
+ Set checked = Set.copyOf(sources);
+ Objects.requireNonNull(message, "message");
+ if (checked.isEmpty()) {
+ throw new IllegalArgumentException("Ready work must reference at least one source");
+ }
+ lock.lock();
+ try {
+ requireReady(Role.PENDING);
+ Entry existing = null;
+ for (SourceId source : checked) {
+ Source accepted = records.get(source);
+ if (accepted == null || !accepted.initiallyHeld) {
+ throw invalidPublication();
+ }
+ if (accepted.ready != null) {
+ if (existing != null && accepted.ready != existing) {
+ throw invalidPublication();
+ }
+ existing = accepted.ready;
+ }
+ }
+ if (existing != null) {
+ if (!existing.sources.equals(checked)) {
+ throw invalidPublication();
+ }
+ for (SourceId source : checked) {
+ if (records.get(source).ready != existing) {
+ throw invalidPublication();
+ }
+ }
+ return;
+ }
+ Entry entry = readyEntry(checked, snapshot(message, Role.PENDING));
+ for (SourceId source : checked) {
+ records.get(source).ready = entry;
+ }
+ addWaiting(entry);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private Entry readyEntry(Set sources, M message) {
+ Instant available;
+ try {
+ available = Objects.requireNonNull(eligibleAt.apply(message), "eligibleAt result");
+ } catch (RuntimeException failure) {
+ throw new OutboundStorageException(Role.PENDING, UNSUPPORTED, "Cannot determine work eligibility", failure);
+ }
+ return new Entry<>(sources, message, available, sequence++);
+ }
+
+ private void addWaiting(Entry entry) {
+ waiting.computeIfAbsent(entry.priority, ignored -> new TreeSet<>(Comparator
+ .comparing((Entry item) -> item.available)
+ .thenComparingLong(item -> item.sequence))).add(entry);
+ }
+
+ private static OutboundStorageException invalidPublication() {
+ return new OutboundStorageException(Role.PENDING, INVALID_TRANSITION,
+ "Ready publication requires retained held sources with one matching source set");
+ }
+
+ /** Single-source lookup for memory-backend consumers; returns an isolated execution value. */
+ public Optional find(SourceId source) {
+ Objects.requireNonNull(source, "source");
+ lock.lock();
+ try {
+ requireReady(Role.PENDING);
+ Source entry = records.get(source);
+ return entry == null ? Optional.empty() : Optional.of(snapshot(entry.message, Role.PENDING));
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public void complete(Set sources) {
+ Set