diff --git a/docs/14-outbound-storage.md b/docs/14-outbound-storage.md
index e9db6c80..cc51dba9 100644
--- a/docs/14-outbound-storage.md
+++ b/docs/14-outbound-storage.md
@@ -6,7 +6,9 @@ Accepted sources remain owned until provider processing and every required track
Only **`memory/memory/memory`** is implemented: unfinished work is lost on process closure or restart.
The contracts are message-neutral; the current standalone provider adapter executes SMPP-client submissions.
-This is the initial slice of [#338](https://github.com/cytechmobile/sendium/issues/338), under [#337](https://github.com/cytechmobile/sendium/issues/337).
+The working lifecycle comes from [#338](https://github.com/cytechmobile/sendium/issues/338), with pending-source
+contracts, multipart admission and the [versioned CBOR representation](15-pending-record-schema.md) refined by
+[#343](https://github.com/cytechmobile/sendium/issues/343), under [#337](https://github.com/cytechmobile/sendium/issues/337).
## Flow Overview
@@ -39,6 +41,8 @@ phases. Arrows show progression, not deletion of earlier records.
multipart parts and in-flight sources, before HTTP 202 / SMPP success. When full, new admissions fail
with **HTTP 503** or **SMPP `STATUS_THROTTLED`**. Existing work continues processing; cleanup frees
space. Moving to a later stage does not free a Pending slot. An aggregate retains every original source ID.
+ Acknowledged duplicate multipart handles remain owned with their original ordinal until terminal completion
+ and required DLR handoff; their payloads are not routed independently.
- **(2) Selected — `SelectedRouterStore`:** holds up to **1,000 queued or taken unassigned items**.
Select up to **100 ready items/cycle**, limited by free slots, and route up to **100 takes/cycle**
through routing rules. When full, additional ready work stays in Pending. Recording a destination
@@ -73,16 +77,34 @@ phases. Arrows show progression, not deletion of earlier records.
`sendium-core` supplies reusable contracts and implementations. `sendium-app` constructs and runs the
standalone pipeline; importing the core library alone does not activate this storage assembly.
+```mermaid
+flowchart LR
+ ingress["HTTP / SMPP admission"] --> coordinator["OutboundCoordinator ownership and transitions"]
+ pipeline["StandaloneOutboundPipeline construction, polling, shutdown"] --> coordinator
+ pipeline --> dispatch["StandardOutboundDispatch routing and provider executions"]
+ dispatch --> coordinator
+ dispatch --> routing["StandardRoutingManager rules and filters"]
+ dispatch --> provider["SmppClientWorker requests and callbacks"]
+ provider --> handoff["Tracker / DlrStorage required handoff"]
+ handoff -->|Completion condition| coordinator
+ coordinator --> pending["PendingMessageStore accepted sources"]
+ coordinator --> selected["SelectedRouterStore selected backlog"]
+ coordinator --> routed["RoutedWorkStore recorded destinations"]
+```
+
| Component | Responsibility |
|---|---|
| `MessageStorageProfile` / `StandaloneMessageStorage` | Validate startup selectors, produce the profile and report its non-durable guarantee. |
| `StandaloneOutboundPipeline` | Start one shared coordinator; drive bounded selection, routing, dispatch, retries and drain-first shutdown. |
| `KannelResource` / `SmppServerWorker` | Admit before HTTP 202 / SMPP success; retain original source identities through multipart preparation. |
| `PendingMessageStore` / `MemoryPendingMessageStore` | Retain accepted sources, held parts and ready work through source completion. |
+| `MultipartKey` / `MultipartMembership` | Match compatible parts and retain each source's assigned group, ordinal, absolute deadline and group creation order. |
+| `PendingSourceRecord` / `PendingMessageData` / `PendingRecordCodec` | Explicit accepted-source representation and versioned Jackson CBOR codec; not a physical backend or automatic memory-store encoding. |
+| `PendingMultipartRecovery` | Shared record-driven assembly/expiry entrypoint using bounded reads; backend/application startup integration is explicit. |
| `SelectedRouterStore` / `MemorySelectedRouterStore` | Own bounded selection and unassigned routing work; retain selected bookkeeping after routing. |
| `RoutedWorkStore` / `MemoryRoutedWorkStore` | Record one destination per selection and schedule its unfinished work by destination. |
| `DefaultOutboundCoordinator` / `OutboundWork` | Enforce typed source/selection/work ownership, transitions, handoff-dependent completion and retryable cleanup. |
-| `StandardRoutingManager` / `StandardOutboundDispatch` | Resolve routing rules, retain recording intents and schedule tracked provider attempts. |
+| `StandardRoutingManager` / `StandardOutboundDispatch` | Resolve rules/filters, retain recording intents and schedule tracked provider attempts. |
| `SmppClientWorker` / `SmppClientSessionHandler` / `CoordinatedSmppSubmission` | Prepare/send requests, process callbacks once and join part/replacement outcomes. |
| `Tracker` / `StandardMessageTracker` | Signal successful required handoff; integrate with the independently configured DLR subsystem. |
@@ -98,13 +120,13 @@ sequenceDiagram
participant P as Provider execution and callbacks
participant H as Required tracking / DLR handoff
Client->>Ingress: Submit message
- Ingress->>C: accept source or acceptHeld part
- C-->>Ingress: Accepted ownership
+ Ingress->>C: accept source or acceptPart with timeout
+ C-->>Ingress: Confirmed source handle / multipart membership
Ingress-->>Client: Success + gateway UUID
D->>C: selectToRouter - select bounded ready work into router backlog
D->>C: Take selected work for routing
C-->>D: Selected payload + original source IDs
- D->>R: Resolve one destination using routing rules
+ D->>R: Resolve one destination and apply routing filters
R-->>D: Destination
D->>C: recordToRouted - record choice and mark selected work routed
D->>C: takeFromRouted - take assigned work for execution
@@ -122,9 +144,10 @@ sequenceDiagram
- **Identity:** HTTP/SMPP reuse each gateway UUID as its typed `SourceId`. `SelectionId` and `WorkId`
identify later steps independently of mutable payloads and provider IDs. Return the actual take handle,
with its updated payload, rather than a reconstructed wrapper; stale returns must not overwrite live work.
-- **Incoming multipart:** `acceptHeld` retains each part before acknowledgement. The existing reassembler
- makes an aggregate or expired individual parts ready using `makeHeldReady(originalSourceIds, payload)`.
- This creates no new admission; publication retries do not re-acknowledge or reassemble the input.
+- **Incoming multipart:** `acceptPart` retains each part and its group membership before acknowledgement.
+ The reassembler makes an aggregate or expired individual parts ready using `makeHeldReady(originalSourceIds, payload)`.
+ This creates no new admission; in-process publication retries do not re-acknowledge input or create
+ another source binding.
- **Outgoing multipart:** provider requests and replacements share one routed parent and original source
set. They are not copied-route branches. No source is removed until all required completion conditions succeed.
- **Completion:** `complete(workId, stage)` can register an outstanding condition; it does not immediately
@@ -133,7 +156,7 @@ sequenceDiagram
- **DLR boundary:** completion waits for required handoff acceptance, not a later handset-delivery receipt.
Receipt storage/delivery remains independent of outbound backend selection.
-`StandardOutboundDispatch.routeSelected()` performs routing lookup.
+`StandardOutboundDispatch.routeSelected()` performs routing lookup and filters.
`OutboundCoordinator.recordToRouted(selected, destination)` records that already-chosen result and
advances stage ownership; it does not choose the destination or send the message.
`takeFromRouted(destination, timeout)` takes one assigned item without completing ownership;
@@ -142,6 +165,73 @@ advances stage ownership; it does not choose the destination or send the message
assembling or accepting new input. `selectToRouter(limit)` then selects eligible work into the router
backlog and returns the newly staged count; it does not evaluate routing rules.
+## Pending Sources and Multipart Groups
+
+`PendingMessageStore.admit` and `admitHeld` return the supplied stable `SourceId` only after acceptance.
+`OutboundCoordinator.accept` and `acceptHeld` propagate that confirmed handle. A null or substituted
+backend result is an ownership failure, never permission to acknowledge untracked work. `acceptPart`
+similarly confirms the source in its returned `MultipartMembership`. `admitHeld` remains available for
+generic held work without assigning it to a multipart group.
+
+`find(source)` returns an isolated copy of the **accepted** payload. It does not return the later routing
+or provider-mutated execution message. `findMultipart(source)` returns immutable membership when present.
+`sourcesAfter(cursor, limit)` enumerates immutable pages of at most `limit` handles in `UUID.compareTo`
+order; start with a null cursor and continue after the last returned handle. Completion of that cursor
+does not invalidate progress. This is a bounded read API, not selection. Recovery callers quiesce ingress
+and processing; concurrent admission/completion is not a consistent snapshot.
+
+Every new client submission receives its own source ID, including a repeated multipart ordinal.
+Only an **internal retry of the same retained source ID** preserves the first accepted value without
+consuming another source slot. A successful earlier client response does not make a later submission
+with the same content/reference a deduplicated internal retry.
+
+Multipart matching uses account/system, sender, recipient, reference width/value, declared total and
+normalized UDH. Other header bytes are preserved, so different ports or headers do not share a group.
+The parser supports one valid 8-bit or 16-bit concatenation element alongside other length-checked IEs.
+Assign the part to the open, unexpired matching group; otherwise create a new group instance.
+The first accepted source for each ordinal wins. Later sources report duplicate membership referencing
+that original source and cannot replace its payload. The first accepted part fixes
+the absolute deadline, and subsequent parts or timeout-configuration changes do not extend it.
+
+For **A=part1, B=part1, C=part2, D=part3**, assemble **A+C+D** and retain B's accepted handle with the
+same group. All four submissions are acknowledged after admission. Completing the aggregate removes
+all four source records after terminal processing and required handoff. B is never routed separately.
+On expiry, each received original part publishes with its duplicate handles. Publication atomically
+includes those handles even when their ingress notifications have not been processed. Mixing group
+instances or publishing an incomplete assigned aggregate is rejected.
+
+The [schema reference](15-pending-record-schema.md) specifies all 54 accepted message fields, DLR return
+metadata, multipart membership, extension types and validation rules. It does not serialize
+`StandardMessage` directly or instantiate application classes from stored names. Normal memory
+operation continues to use its existing snapshot policy; the standalone CBOR mapper is an explicit
+library contract for backends, not a new serialization step on every memory admission.
+
+### Record-driven reconstruction
+
+`PendingMultipartRecovery` rebuilds runtime assembly from accepted records, saved assignments and
+absolute deadlines. It validates bounded source pages and group-handle indexes before reporting ready.
+Complete retained groups assemble in ordinal order; expired incomplete groups release received parts;
+unexpired incomplete groups schedule only the time remaining. A restart at 12:00:50 with a saved
+12:01:00 deadline leaves ten seconds, not another full timeout.
+
+Waiting groups retain handles/metadata, not a second payload backlog. `multipartSources(groupId)` reads
+one group's retained handles, including duplicates and parts still awaiting ingress notification,
+bounded by pending source capacity. There are at most 255 distinct ordinals.
+Payloads are fetched for assembly or expiry. Closing recovery cancels its timers; source completion
+still belongs to the coordinator after terminal processing and required DLR handoff.
+
+There is **no new durable checkpoint of a runtime assembly, release or publication decision**.
+Pending-only recovery may repeat processing. Selection/destination decisions survive only under
+their own configured stage guarantees. Within a running process, `makeHeldReady` retains the first
+source-to-projection binding and prepared payload, making publication retries idempotent.
+
+`SmppServerWorker.reconstructPendingParts(pending, pageSize)` is an explicit pre-ingress startup hook
+using its selected coordinator's publication boundary. It prevents worker startup/admission success
+after failed reconstruction. It is **not** automatic restart recovery in the current standalone memory
+profile. Opening/indexing a physical durable backend, invoking reconstruction before ingress/dispatch,
+reconciling later stages and enabling a supported durable profile remain backend/application work in
+#339 and later issues. The current coordinator still rejects durable stages.
+
## Ownership and Capacity
The stores are process-local objects, not three independent batches. The routed store contains queues
@@ -171,12 +261,14 @@ A group of N accepted multipart sources uses N pending slots but one selected sl
Partial terminal cleanup can retain routed records after source removal, so equal capacities do not
eliminate cleanup-related backpressure or provide cross-store transactions.
-Selection or returning an unroutable item alone does not count as productive cycle progress.
-Shutdown prevents further cycle scheduling. Batch size is not TPS or provider
+The processor schedules the next bounded cycle without a mandatory delay after successful routing
+or provider scheduling. Idle, paused, or unsuccessful cycles back off for 100 milliseconds;
+selection or returning an unroutable item alone does not count as progress. Shutdown prevents
+further cycle scheduling. Batch size is not TPS or provider
concurrency: attempts use the worker's shared rate limiter and a dispatcher-owned execution pool
sized by its `threadCount`. At most that many preparation loops execute concurrently, taking waiting
attempts from the worker FIFO. The default shared two-thread scheduler handles destination resolution
-and timed retries; rate waits and SMPP sends run on the worker's pool.
+and timed retries; rate waits, worker filters, and SMPP sends run on the worker's pool.
Stopped-worker pools are retired during transition maintenance, and dispatcher close shuts down
its worker pools after provider drain even when the timer scheduler is application-owned.
Awaiting asynchronous outcomes does not hold a sending slot. Count limits do not bound message bytes.
@@ -210,12 +302,15 @@ attempts, not either storage-return operation.
| Admission capacity/unavailability | HTTP 503; SMPP capacity returns `STATUS_THROTTLED`. Failed admission is not acknowledged as success. |
| Routing miss / nonterminal lookup failure | Use `requeueForRouting` with ownership retained. A terminal routing drop uses `discard`. |
| Copied route / unsupported standalone destination | Reject before assignment, even if a copied rule resolves to one target. The existing legacy copied-routing API remains separate. |
-| Destination-recording failure | Retry the retained original intent without repeating routing lookup. |
+| Destination-recording failure | Retry the retained original intent without repeating lookup or routing filters. |
| Required handoff failure | Retain ownership and retry the handoff, not an accepted provider request. |
| Terminal cleanup failure | Retry unfinished cleanup without awaiting another successful handoff or redispatching provider work. |
+| Uncertain source absence | Completion fails explicitly; only established absence on a ready backend permits idempotent success. |
+| Corrupt/unknown-version recovery input | Fail reconstruction/readiness explicitly; do not skip the source or treat it as completed. |
Each provider response is claimed once; duplicate callbacks cannot create another replacement or
-complete unrelated sources. Admission deduplication applies while ownership is retained, not forever.
+complete unrelated sources. Same-source-ID admission idempotence applies while ownership is retained,
+not forever; new client submissions remain independent.
Runtime worker removal's legacy queue return does not migrate coordinator-owned parent records.
Provider preparation uses a per-worker FIFO of already-owned runtime attempts. Up to the worker's
@@ -288,8 +383,17 @@ existing router/worker startup. A failing coordinated admission never falls back
Future profiles are not selectable today. Memory remains the non-durable default; planned file profiles
are opt-in. Durable recovery must reconcile identities and resume the latest recorded stage without
scheduling earlier retained records independently. It requires surviving storage and remains at least
-once: provider submission before recorded completion may repeat. Physical backends, codecs and durable
-provider-outcome checkpoints belong to subsequent issues.
+once: provider submission before recorded completion may repeat. The CBOR source codec is implemented;
+physical backends and durable later-stage implementations belong to subsequent issues. Durable provider
+outcome/retry checkpoints remain outside this delivery.
+
+### Delivery and rollback boundary
+
+#343 adds pending contracts, source representation, shared runtime reconstruction and behavioral proofs.
+It adds no spool directory/publication scheme, PostgreSQL outbound schema, durable profile or durable-data
+migration. Its dependent commits can be reverted together, in reverse order, to the working #338 memory
+baseline while retaining that baseline's performance fixes. This code-only rollback does not recover
+unfinished memory work across a restart. Existing PostgreSQL DLR storage remains independently owned.
## Embedding and API Reference
@@ -306,6 +410,8 @@ Detailed operation, idempotence and failure rules are defined in the tracked Jav
- [OutboundCoordinator](../sendium-core/src/main/java/gr/cytech/sendium/core/outbound/OutboundCoordinator.java) and [OutboundWork](../sendium-core/src/main/java/gr/cytech/sendium/core/outbound/OutboundWork.java)
- [PendingMessageStore](../sendium-core/src/main/java/gr/cytech/sendium/core/storage/PendingMessageStore.java)
+- [PendingRecordCodec](../sendium-core/src/main/java/gr/cytech/sendium/core/storage/pending/PendingRecordCodec.java) and [pending record schema](15-pending-record-schema.md)
+- [PendingMultipartRecovery](../sendium-core/src/main/java/gr/cytech/sendium/core/storage/pending/PendingMultipartRecovery.java)
- [SelectedRouterStore](../sendium-core/src/main/java/gr/cytech/sendium/core/storage/SelectedRouterStore.java)
- [RoutedWorkStore](../sendium-core/src/main/java/gr/cytech/sendium/core/storage/RoutedWorkStore.java)
- [OutboundStage](../sendium-core/src/main/java/gr/cytech/sendium/core/storage/OutboundStage.java) and [OutboundStorageException](../sendium-core/src/main/java/gr/cytech/sendium/core/storage/OutboundStorageException.java)
@@ -316,3 +422,9 @@ shutdown ordering and startup rejection. Provider transport/handoffs use control
external provider. Key tests are [StandaloneOutboundPipelineTest](../sendium-app/src/test/java/gr/cytech/sendium/app/storage/StandaloneOutboundPipelineTest.java),
[SmppClientWorkerTest](../sendium-core/src/test/java/gr/cytech/sendium/core/smpp/client/SmppClientWorkerTest.java) and
[OutboundEmbeddedLifecycleTest](../sendium-core/src/test/java/gr/cytech/sendium/core/outbound/consumer/OutboundEmbeddedLifecycleTest.java).
+Pending-specific proofs are [MemoryPendingMessageStoreContractTest](../sendium-core/src/test/java/gr/cytech/sendium/core/storage/pending/MemoryPendingMessageStoreContractTest.java),
+[PendingRecordMappingTest](../sendium-core/src/test/java/gr/cytech/sendium/core/storage/pending/PendingRecordMappingTest.java),
+[PendingRecordCodecTest](../sendium-core/src/test/java/gr/cytech/sendium/core/storage/pending/PendingRecordCodecTest.java),
+[PendingCompletionFailureTest](../sendium-core/src/test/java/gr/cytech/sendium/core/storage/pending/PendingCompletionFailureTest.java) and
+[PendingMultipartRecoveryTest](../sendium-core/src/test/java/gr/cytech/sendium/core/storage/pending/PendingMultipartRecoveryTest.java).
+Codec-backed reconstruction fixtures are not physical file-publication or forced-process restart tests.
diff --git a/docs/15-pending-record-schema.md b/docs/15-pending-record-schema.md
new file mode 100644
index 00000000..f1201a70
--- /dev/null
+++ b/docs/15-pending-record-schema.md
@@ -0,0 +1,314 @@
+# Pending accepted-source record — CBOR schema
+
+[Outbound storage lifecycle](14-outbound-storage.md) · [Documentation map](DocumentationMap.md)
+
+This documents the explicit pending-source representation introduced by #343.
+It is not Java serialization, a serialization of `StandardMessage`, a file layout,
+or a supported durable backend. The standalone runtime still supports only
+`memory/memory/memory`. Filesystem admission/restart recovery belongs to #339.
+
+## API and ownership
+
+`PendingSourceRecord` contains the existing typed `OutboundWork.SourceId`, an
+accepted disposition (`READY` or `HELD`), `PendingMessageData`, and optional
+`MultipartMembership`. The disposition is the original admission disposition,
+not a checkpoint of subsequent router/routed ownership.
+`PendingMessageData.capture(StandardMessage)` captures the accepted values using
+an explicit mapping; `restore()` returns a new mutable runtime message each time.
+Maps, lists and byte arrays are detached from both input and restored messages.
+The source ID must equal the gateway UUID in `serial` (canonical UUID syntax,
+case-insensitive). Custom `StandardMessage` subclasses are not supported by this
+mapper. No existing memory snapshot or ingress behavior is changed by the codec.
+
+`PendingRecordCodec.encode(record)` returns CBOR bytes.
+`PendingRecordCodec.decode(bytes)` validates one complete envelope and returns a
+record. The codec has no storage, acknowledgement, routing, or lifecycle side
+effects. Later admission integration must fail before client success if a selected
+durable representation cannot preserve the accepted data. No codec integration
+or new backend profile is enabled by the record/codec work.
+
+## Version 1 envelope
+
+The top-level value is a CBOR map with text keys, emitted in this order:
+
+| Field | Type | Required meaning |
+|---|---|---|
+| `schemaVersion` | Positive signed 32-bit integer | `1` |
+| `sourceId` | UUID text | Existing gateway/source UUID |
+| `disposition` | Text enum | `READY` or `HELD` |
+| `message` | Map | Explicit message fields below |
+| `multipart` | Null or map | Recorded multipart membership below; null for ordinary sources |
+
+All documented fields are present, including fields whose value is null.
+Missing fields are corruption, not an instruction to use Java constructor
+defaults. Null is permitted for message text/reference/container fields as listed
+below; it is not interchangeable with empty text, an empty collection, zero or
+false. Primitive fields cannot be null. Numeric range/type checks do not coerce
+strings, booleans or fractional numbers into integer fields. `dcs` uses the signed
+Java byte representation (-128 through 127), preserving all eight bits.
+
+Version 1 is the explicit accepted-source schema implemented by #343. The earlier
+complete-source development commit was not a released durable format. The
+`multipart` key is required (null for ordinary sources). Intermediate development
+records without this key or the required membership `originalSourceId` are rejected, rather than silently interpreted as a complete
+durable schema. Saved membership does not itself implement process-restart recovery;
+shared reconstruction is described below, and a physical backend follows in #339.
+
+## Multipart admission and membership
+
+`PendingMessageStore.admitPart(source, message, timeout)` atomically retains the
+held source and assigns its group before returning `MultipartMembership`.
+`OutboundCoordinator.acceptPart` exposes that confirmed result to SMPP ingress;
+`findMultipart(source)` retrieves the immutable membership through the pending port.
+Plain `admitHeld` remains available for held sources without a multipart assignment.
+
+Each **new source ID** is accepted and retained. Join the open, unexpired matching
+group or create a fresh group UUID. The first accepted source for an ordinal wins;
+later sources reference it through `originalSourceId` and report `duplicate() == true`.
+Their payloads never replace the original or route independently. The saved `groupOrder` is carried in every member record,
+so a later recovery does not have to guess the order of groups with equal deadlines.
+Internal retries using the same retained source ID return
+its original membership and never overwrite its accepted payload. A source admitted
+without multipart membership cannot be converted into one by retrying `admitPart`.
+
+The matching key has account ID, system ID, sender, recipient, reference width/value,
+declared total and normalized UDH. The parser checks the entire hex-encoded header:
+UDHL must match its byte count; each information element must fit; exactly one
+valid 8-bit or 16-bit concatenation element must exist; total is 1–255 and ordinal
+is 1–total. Concatenation may appear before or after other IEs. Normalization uses
+uppercase hex and replaces only the concatenation ordinal byte with zero. All
+other bytes, including IE order and port information, are preserved; reordered IEs
+are not asserted equivalent by this representation.
+
+The first accepted part fixes an absolute deadline using the backend's clock and
+the configured timeout. Later parts do not extend it, even if the configured timeout
+changes. Expiry is inclusive: at the deadline, the group cannot receive new parts.
+Complete groups, groups published for processing, and groups whose sources begin
+completion also stop accepting new members. This does not remove retained sources
+or their saved membership; source removal still requires terminal processing and
+the required DLR handoff.
+
+Example: A=part1, B=part1, C=part2, D=part3 produce one group with four accepted handles.
+Assembly uses A+C+D; B's membership references A. Terminal completion includes A/B/C/D.
+If incomplete at expiry, A publishes with B's handle and the other received original
+ordinals publish independently. Publication includes duplicate handles atomically, even
+before their ingress notifications are processed. Existing non-coordinated legacy
+entrypoints retain their first-ordinal-wins behavior.
+
+| Membership field | Stored type / meaning |
+|---|---|
+| `sourceId` | Canonical UUID text; original part UUID, must match envelope/source serial |
+| `originalSourceId` | Canonical UUID text; first accepted source for this ordinal; equals `sourceId` for an original, differs for a duplicate |
+| `groupId` | Canonical UUID text; stable assigned group instance |
+| `ordinal` | Integer in 1–`key.total` |
+| `deadline` | ISO-8601 UTC `Instant` text, preserving nanoseconds |
+| `groupOrder` | Non-negative int64; stable group creation order within the backend's retained history |
+| `key` | Explicit identity map below |
+
+The key map contains required nullable text fields `accountId`, `systemId`, `sender`
+and `recipient`; integer `referenceBits` (8 or 16), `reference` (0–255 or 0–65535
+respectively), `total` (1–255); and required non-null text `normalizedUdh`.
+Null remains distinct from empty text for matching identity. Non-null membership
+requires `HELD` original disposition. Its key and ordinal must match the accepted
+payload's parsed UDH and addresses; mismatches are corruption, not a request to
+reassign the source on decode. Decode never calculates a new group UUID or deadline.
+
+## Reconstructing runtime assembly
+
+`PendingMultipartRecovery` reads a quiescent pending store through bounded
+`sourcesAfter` pages. It indexes only source handles and multipart metadata,
+validating the complete scan before publishing work or declaring recovery ready.
+Group members must agree on their key, group UUID, creation order and deadline.
+Each ordinal has one original source; duplicates must reference that retained original
+in the same group and ordinal. Missing or conflicting originals fail reconstruction.
+
+- A complete retained group is assembled in ordinal order using detached accepted
+ payloads, retaining the original source IDs for completion and DLR handling.
+- An incomplete expired group releases its original parts independently, each owning its duplicate handles.
+- An incomplete unexpired group schedules its original deadline. Reconstructing
+ at 12:00:50 with deadline 12:01:00 leaves ten seconds, regardless of downtime.
+- `multipartSources(groupId)` returns only that group's retained handles, including
+ duplicates, bounded by pending source capacity. It includes accepted parts whose ingress
+ notifications have not been processed. Payloads are fetched only on assembly or
+ expiry; there is no second cache containing the entire backlog's message bodies.
+- A new part notification carries a source handle. Current membership and payload
+ remain authoritative in pending storage, including after reconstruction.
+
+There is **no durable checkpoint of an earlier assembly/publication/release
+decision**. Accepted-source CBOR, membership and deadlines are the reconstruction
+inputs. Pending-only recovery can repeat runtime processing; selected batches and
+destinations survive only under their independently configured stage guarantees.
+Publication retries in a running process use the existing first-binding/idempotence
+contract of `makeHeldReady`. Transient timer-publication failures retry after one
+second without accepting new sources or making provider attempts. Corrupt recovery
+input fails explicitly, leaving accepted ownership intact.
+
+`SmppServerWorker.reconstructPendingParts(pending, pageSize)` is the startup
+entrypoint after selecting its coordinator and opening the owning pending store,
+before starting ingress/dispatch or accepting submissions. It uses the existing
+coordinator publication boundary. Failed reconstruction prevents worker startup
+and success acknowledgements; it does not fall back to legacy processing. Existing
+non-record-driven entrypoints remain supported. Reassembly timeout reconfiguration
+affects new admissions, not saved group deadlines or the active recovery component.
+
+The application owns the scheduler and pending store; closing the recovery component
+cancels its own timers only. The handler owns and stops its scheduler as before.
+Actual durable backend opening, accepted-record indexing, stage reconciliation and
+profile enablement remain with the backend/application assembly in #339 and later
+stage implementations. This entrypoint does not remove the coordinator's current
+guard rejecting unsupported durable profiles. The existing memory backend still
+loses unfinished records when closed or restarted.
+
+Recovery tests use copied codec-backed source records and a controlled clock/timer.
+They verify reconstruction, not physical file publication or forced-process restart
+durability. Full group payloads may be loaded while assembling one message (at most
+255 parts); metadata indices can grow with the count of incomplete groups. Bounded
+recovery access is not a claim of constant memory for those indices.
+
+## Explicit message field inventory
+
+Every public **instance** field of `StandardMessage` is preserved at capture time.
+None is implicitly omitted, derived from a new wall-clock time, or obtained by
+reflective message serialization. Fields that later change during routing or
+provider processing retain their **accepted** value here: this is not a checkpoint
+of later runtime execution. Constants, `fieldMap`, the logger, methods and Java
+class identity are not message data and are not encoded.
+
+The parameterized `PendingRecordMappingTest#roundTripsEveryPreservedField` discovers
+public instance fields for test coverage only. A new field without an explicit
+capture/restore mapping fails the test instead of silently inheriting serialization.
+
+| Runtime field | Schema field | Stored type / reason |
+|---|---|---|
+| `from` | `from` | Nullable text; sender |
+| `to` | `to` | Nullable text; recipient |
+| `timestamp` | `timestamp` | Nullable text; original timestamp representation |
+| `type` | `type` | Int32; message type |
+| `body` | `body` | Nullable text; accepted body |
+| `serial` | `serial` | UUID text; gateway identity |
+| `message_center` | `messageCenter` | Nullable text; message center/input |
+| `owner_id` | `ownerId` | Nullable text; account identity |
+| `systemId` | `systemId` | Nullable text; system identity |
+| `priority` | `priority` | Int32; selection/routing input |
+| `acked` | `acked` | Boolean; accepted acknowledgement-related value |
+| `msgId` | `msgId` | Int32; accepted message field |
+| `expr` | `expr` | Int64; existing eligibility/order input |
+| `rtxCnt` | `rtxCnt` | Int32; initial accepted value, not a durable retry checkpoint |
+| `binbody` | `binbody` | Nullable text; unchanged existing binary-body representation |
+| `binheader` | `binheader` | Nullable text; unchanged UDH representation |
+| `flash` | `flash` | Boolean; message formatting |
+| `mclass` | `mclass` | Int32; message class |
+| `ttl` | `ttl` | Int32; validity input |
+| `ddt` | `ddt` | Int32; deferred-delivery input |
+| `dcs` | `dcs` | Signed int8; data coding |
+| `onetwork` | `onetwork` | Int32; original network/routing input |
+| `cnetwork` | `cnetwork` | Int32; current network at acceptance |
+| `ingateway` | `ingateway` | Nullable text; ingress identity |
+| `outgateway` | `outgateway` | Nullable text; accepted routing/worker input |
+| `state` | `state` | Int32; accepted state, not a provider-outcome checkpoint |
+| `cost` | `cost` | Int64; accepted processing input |
+| `ctstamp` | `ctstamp` | Int64; accepted creation timestamp |
+| `sutstamp` | `sutstamp` | Int64; accepted submit timestamp |
+| `errcode` | `errcode` | Nullable text; accepted error-related value |
+| `extrid` | `extrid` | Nullable text; external identity |
+| `nextTarget` | `nextTarget` | Nullable text; routing input |
+| `field1` | `field1` | Tagged extension; includes normal HTTP protocol ID input |
+| `field2` | `field2` | Tagged extension; includes normal HTTP alternate DCS input |
+| `field3` | `field3` | Tagged extension; includes normal HTTP reply-path input |
+| `field4` | `field4` | Tagged extension; includes normal HTTP billing-info input |
+| `field5` | `field5` | Tagged extension; accepted input only |
+| `field6` | `field6` | Tagged extension; accepted input only |
+| `field7` | `field7` | Tagged extension; accepted input only |
+| `field8` | `field8` | Tagged extension; accepted input only |
+| `field9` | `field9` | Tagged extension; accepted input only |
+| `field10` | `field10` | Tagged extension; no later filter-phase marker synthesized |
+| `paid` | `paid` | Boolean; accepted processing input |
+| `metadata` | `metadata` | Nullable text; opaque metadata, not parsed/reformatted |
+| `smsCnt` | `smsCnt` | Int32; initial accepted count |
+| `smsSubmitCnt` | `smsSubmitCnt` | Int32; initial accepted count |
+| `attrs` | `attrs` | Null or tagged string-keyed map; DLR/request and processing inputs |
+| `reassembledParts` | `reassembledParts` | Null or array of nullable text; original source identities if present |
+| `dlrReturnMetadata` | `dlrReturnMetadata` | Null or explicit DLR map below, despite runtime `@JsonIgnore` |
+| `tlvs` | `tlvs` | Null or sorted text-keyed map of nullable text; SMPP TLVs |
+| `dispatchType` | `dispatchType` | Int32; dispatch input |
+| `hlrRoute` | `hlrRoute` | Nullable text; routing input |
+| `marginPercentage` | `marginPercentage` | Finite float64; accepted processing input |
+| `marginStatus` | `marginStatus` | Null or named known enum (`NOT_CALCULATED`, `KNOWN`, `UNKNOWN`) |
+
+Preserving an initial count, state or routing field does not authorize persisting
+later retries/outcomes or routing-time filter mutations as source state. Recovering
+only the pending stage reruns selection/routing, as defined by the epic.
+
+### DLR return map
+
+`channel` is required non-null enum text (`HTTP` or `SMPP`). `accountId`, `systemId`,
+`sourceAddress`, `destinationAddress` and `forwardDlrUrl` are required keys with
+nullable text values. They preserve the corresponding `DlrReturnMetadata` record
+components verbatim. Request flags in the message's normal protocol fields/attrs
+are preserved by the message mapping; no DLR backend is serialized here.
+
+## Extension vocabulary
+
+An extension is either CBOR null, or a map containing `type` and `value`:
+
+| `type` | `value` | Restored Java value |
+|---|---|---|
+| `string` | Non-null text | `String` |
+| `boolean` | Boolean | `Boolean` |
+| `int8` | Integer in signed 8-bit range | `Byte` |
+| `int16` | Integer in signed 16-bit range | `Short` |
+| `int32` | Integer in signed 32-bit range | `Integer` |
+| `int64` | Integer in signed 64-bit range | `Long` |
+| `float32` | Finite number exactly representable as float32 | `Float` |
+| `float64` | Finite number | `Double` |
+| `bytes` | Native CBOR byte string, not base64 text | Detached `byte[]` |
+| `list` | Array of extensions | Detached `ArrayList` |
+| `map` | Text-keyed map of extensions | Detached map |
+
+Map keys are emitted in Java `String` natural order (at every nesting level).
+Array order is retained. Scalar tags distinguish, for example, integer `1` from
+long `1L` even if CBOR's preferred integer representation uses the same width.
+This is deterministic encoding by this schema/codec, not a claim of implementing
+every RFC canonical-CBOR option or preserving original collection classes.
+
+Example in diagnostic JSON notation (actual bytes are CBOR):
+
+```json
+{"type":"map","value":{"attemptLimit":{"type":"int32","value":3},"campaign":{"type":"string","value":"summer"}}}
+```
+
+Unsupported values include arbitrary application objects/enums, characters,
+non-string map keys, other array types, NaN/infinities, cyclic containers and
+nesting beyond 64 container levels (root is level 0; levels 0–64 are supported).
+Shared acyclic containers may occur in several places and are copied as values.
+The codec rejects unsupported data explicitly; it never substitutes `toString()`
+or skips an unsupported entry. Custom before-insert filter object support is
+deferred. Existing in-memory snapshot rules remain unchanged.
+
+## Validation, compatibility and failures
+
+- Positive unknown schema versions produce `PendingRecordException` with
+ `kind() == UNSUPPORTED_VERSION` and the existing storage reason `UNSUPPORTED`.
+- Malformed/truncated CBOR, missing required fields, invalid primitive types or
+ ranges, unknown enum/type tags, mismatched gateway/source UUIDs, duplicate map
+ keys, invalid extension values and trailing CBOR values produce
+ `kind() == CORRUPT_RECORD` with storage reason `UNAVAILABLE`. A corrupt source
+ cannot be treated as absent. These codec-specific details preserve the existing
+ protocol-facing storage reason vocabulary.
+- Encode/capture of unsupported application values produces storage reason
+ `UNSUPPORTED`. All codec errors identify pending stage and codec context, with
+ fixed diagnostics and no parser-cause chain containing payload values.
+- Unknown optional keys within version 1 are ignored. They cannot change the
+ meaning of required fields; semantic changes require a new version. Unknown
+ fields are not promised to survive decode/re-encode.
+- No Java class names are interpreted, no Jackson default typing is enabled,
+ and the runtime `StandardMessage` JSON contract is untouched.
+- The parser limits total CBOR nesting to 256, sufficient for this tagged
+ extension vocabulary's 64-level limit and envelope overhead.
+- There is no earlier durable pending schema to migrate in this delivery.
+ Backward migration/version adapters are not implemented speculatively.
+
+The codec detects structural and semantic corruption. It is not a checksum or
+an authenticated format: arbitrary bit changes that still form valid supported
+values cannot all be detected by the codec. File publication/integrity policy is
+owned by the physical backend implementation.
diff --git a/docs/DocumentationMap.md b/docs/DocumentationMap.md
index 8c74c4db..ee295bcb 100644
--- a/docs/DocumentationMap.md
+++ b/docs/DocumentationMap.md
@@ -23,6 +23,7 @@ Sendium is an open-source, headless SMS gateway for high-throughput messaging. I
| Review current features and roadmap | [12. Features And Roadmap](12-features-roadmap.md) |
| Configure PostgreSQL DLR persistence and review durability | [13. DLR Persistence](13-dlr-persistence.md) |
| Understand outbound storage and extension contracts | [14. Outbound Message Storage](14-outbound-storage.md) |
+| Implement pending-source encoding and reconstruction | [15. Pending Record Schema](15-pending-record-schema.md) |
| Contribute code or docs | [Contributing](../.github/CONTRIBUTING.md) |
## Core Concepts
@@ -65,6 +66,7 @@ Sendium expects these files in the configured `conf` directory. The Docker quick
| [12. Features And Roadmap](12-features-roadmap.md) | Current product capabilities, planned roadmap phases, and related feature documentation. |
| [13. DLR Persistence](13-dlr-persistence.md) | PostgreSQL setup, retention, restart guarantees, and durability limits. |
| [14. Outbound Message Storage](14-outbound-storage.md) | Diagram-led lifecycle guide, class responsibilities, capacity bounds, retries, shutdown, profiles, and extension contracts. |
+| [15. Pending Record Schema](15-pending-record-schema.md) | Versioned CBOR source fields, multipart membership/deadlines, supported extensions, validation and record-driven reconstruction boundaries. |
| [Kannel migration converter](https://cytechmobile.github.io/sendium/kannelconverter/) | Browser-only helper for turning a legacy `kannel.conf` into Sendium starter files. |
## API Discovery
@@ -77,7 +79,7 @@ When Sendium is running, the HTTP API can be inspected through:
| `/swagger-ui` | Interactive Swagger UI. |
| `/openapi.json` | OpenAPI specification. |
| `/q/metrics` | Prometheus-compatible Micrometer metrics endpoint. |
-| `/q/health/ready` | Readiness status, including PostgreSQL DLR storage availability. |
+| `/q/health/ready` | Readiness status, including configured DLR storage and the standalone outbound pipeline. |
## Community And Project Files
diff --git a/sendium-app/src/test/java/gr/cytech/sendium/app/storage/MessageLifecycleHttpTest.java b/sendium-app/src/test/java/gr/cytech/sendium/app/storage/MessageLifecycleHttpTest.java
index 6d37bc9a..3fca5a83 100644
--- a/sendium-app/src/test/java/gr/cytech/sendium/app/storage/MessageLifecycleHttpTest.java
+++ b/sendium-app/src/test/java/gr/cytech/sendium/app/storage/MessageLifecycleHttpTest.java
@@ -7,7 +7,11 @@
import gr.cytech.sendium.core.message.StandardMessage;
import gr.cytech.sendium.core.outbound.DefaultOutboundCoordinator;
import gr.cytech.sendium.core.outbound.OutboundCoordinator;
+import gr.cytech.sendium.core.outbound.OutboundWork.SourceId;
import gr.cytech.sendium.core.queue.InMemoryQueueProvider;
+import gr.cytech.sendium.core.storage.OutboundStage;
+import gr.cytech.sendium.core.storage.OutboundStorageException;
+import gr.cytech.sendium.core.storage.PendingMessageStore;
import gr.cytech.sendium.core.storage.memory.MemoryPendingMessageStore;
import gr.cytech.sendium.core.storage.memory.MemoryRoutedWorkStore;
import gr.cytech.sendium.core.storage.memory.MemorySelectedRouterStore;
@@ -19,7 +23,10 @@
import jakarta.inject.Inject;
import jakarta.inject.Singleton;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.net.URI;
import java.net.http.HttpClient;
@@ -32,6 +39,11 @@
import java.util.concurrent.CompletableFuture;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.spy;
class MessageLifecycleHttpTest {
@RegisterExtension
@@ -52,14 +64,25 @@ class MessageLifecycleHttpTest {
@Inject
InMemoryQueueProvider legacy;
+ @Inject
+ PendingMessageStore pending;
+
+ @AfterEach
+ void clearAdmissionFaults() {
+ reset(pending);
+ }
+
@Test
- void httpUsesTheExplicitCoordinatorBeanAndReturnsBackpressureAtCapacity() throws Exception {
+ void successReferencesAcceptedHandle() throws Exception {
URI requestUri = URI.create(endpoint + "?username=http-user&password=secret&from=Sender&to=306900000001&text=Hello" +
"&dlr-url=https%3A%2F%2Fcallback.test%2Fdlr");
try (HttpClient client = HttpClient.newHttpClient()) {
var request = HttpRequest.newBuilder(requestUri).timeout(Duration.ofSeconds(5)).GET().build();
var accepted = client.send(request, HttpResponse.BodyHandlers.ofString());
assertThat(accepted.statusCode()).isEqualTo(202);
+ var retained = pending.find(new SourceId(UUID.fromString(accepted.body()))).orElseThrow();
+ assertThat(retained.serial).isEqualTo(accepted.body());
+ assertThat(retained.body).isEqualTo("Hello");
var full = client.send(request, HttpResponse.BodyHandlers.ofString());
assertThat(full.statusCode()).isEqualTo(503);
assertThat(legacy.getRouterQueue().isEmpty()).isTrue();
@@ -75,12 +98,54 @@ void httpUsesTheExplicitCoordinatorBeanAndReturnsBackpressureAtCapacity() throws
}
}
+ @ParameterizedTest
+ @ValueSource(strings = {"UNAVAILABLE", "CAPACITY_EXCEEDED", "OWNERSHIP_CONFLICT", "UNSUPPORTED", "INVALID_TRANSITION"})
+ void rejectsPendingAdmissionFailures(String reasonName) throws Exception {
+ // Strings cross QuarkusExtensionTest's class-loader boundary; application enum instances do not.
+ var reason = OutboundStorageException.Reason.valueOf(reasonName);
+ doThrow(new OutboundStorageException(OutboundStage.Role.PENDING, reason, "private storage details"))
+ .when(pending).admit(any(), any());
+ try (HttpClient client = HttpClient.newHttpClient()) {
+ var request = HttpRequest.newBuilder(URI.create(endpoint +
+ "?username=http-user&password=secret&from=Sender&to=306900000001&text=Hello"))
+ .timeout(Duration.ofSeconds(5)).GET().build();
+ var response = client.send(request, HttpResponse.BodyHandlers.ofString());
+ int expected = switch (reason) {
+ case UNSUPPORTED -> 400;
+ case INVALID_TRANSITION -> 500;
+ default -> 503;
+ };
+ assertThat(response.statusCode()).isEqualTo(expected);
+ assertThat(response.body()).doesNotContain("private storage details");
+ assertThat(pending.sourcesAfter(null, 1)).isEmpty();
+ assertThat(legacy.getRouterQueue().isEmpty()).isTrue();
+ }
+ }
+
+ @Test
+ void unconfirmedPendingHandleIsNeverAcknowledged() throws Exception {
+ doReturn(null).when(pending).admit(any(), any());
+ try (HttpClient client = HttpClient.newHttpClient()) {
+ var request = HttpRequest.newBuilder(URI.create(endpoint +
+ "?username=http-user&password=secret&from=Sender&to=306900000001&text=Hello"))
+ .timeout(Duration.ofSeconds(5)).GET().build();
+ assertThat(client.send(request, HttpResponse.BodyHandlers.ofString()).statusCode()).isEqualTo(503);
+ assertThat(pending.sourcesAfter(null, 1)).isEmpty();
+ assertThat(legacy.getRouterQueue().isEmpty()).isTrue();
+ }
+ }
+
@Singleton
public static class Beans {
@Produces
@Singleton
- OutboundCoordinator coordinator() {
- var pending = new MemoryPendingMessageStore(1, Beans::snapshot);
+ MemoryPendingMessageStore pending() {
+ return spy(new MemoryPendingMessageStore(1, Beans::snapshot));
+ }
+
+ @Produces
+ @Singleton
+ OutboundCoordinator coordinator(MemoryPendingMessageStore pending) {
var router = new MemorySelectedRouterStore<>(pending, 1);
var routed = new MemoryRoutedWorkStore(1, Beans::snapshot);
var coordinator = new DefaultOutboundCoordinator<>(pending, router, routed);
diff --git a/sendium-app/src/test/java/gr/cytech/sendium/app/storage/StandaloneMessageSnapshotTest.java b/sendium-app/src/test/java/gr/cytech/sendium/app/storage/StandaloneMessageSnapshotTest.java
index 89349250..a5659404 100644
--- a/sendium-app/src/test/java/gr/cytech/sendium/app/storage/StandaloneMessageSnapshotTest.java
+++ b/sendium-app/src/test/java/gr/cytech/sendium/app/storage/StandaloneMessageSnapshotTest.java
@@ -1,16 +1,73 @@
package gr.cytech.sendium.app.storage;
+import gr.cytech.sendium.core.message.DlrReturnMetadata;
import gr.cytech.sendium.core.message.StandardMessage;
+import gr.cytech.sendium.core.outbound.OutboundWork.SourceId;
+import gr.cytech.sendium.core.storage.PendingMessageStore;
+import gr.cytech.sendium.core.storage.memory.MemoryPendingMessageStore;
+import gr.cytech.sendium.core.storage.pending.PendingMessageData;
+import gr.cytech.sendium.core.storage.pending.PendingRecordCodec;
+import gr.cytech.sendium.core.storage.pending.PendingSourceRecord;
import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
class StandaloneMessageSnapshotTest {
+ @Test
+ void normalProtocolRecordPreservesProcessingInputs() {
+ for (var metadata : List.of(DlrReturnMetadata.http("account", "sender", "recipient", "https://callback.test/dlr"),
+ DlrReturnMetadata.smpp("account", "system", "sender", "recipient"))) {
+ var message = new StandardMessage();
+ SourceId source = new SourceId(UUID.randomUUID());
+ message.serial = source.value().toString();
+ message.owner_id = "account";
+ message.systemId = "system";
+ message.from = "sender";
+ message.to = "recipient";
+ message.body = "Hello";
+ message.priority = 3;
+ message.ttl = 120;
+ message.ddt = 5;
+ message.dcs = StandardMessage.DCS_16BIT;
+ message.nextTarget = "provider";
+ message.field1 = 7;
+ message.field2 = 8;
+ message.field3 = 1;
+ message.field4 = "billing-info";
+ message.acked = true;
+ message.dlrReturnMetadata = metadata;
+ message.tlvs = new HashMap<>(Map.of("0427", "1"));
+ message.attrs = new HashMap<>(Map.of("request", "original"));
+
+ try (PendingMessageStore pending = new MemoryPendingMessageStore<>(1, StandaloneMessageSnapshot::copy)) {
+ pending.open();
+ SourceId accepted = pending.admit(source, message);
+ var retained = pending.find(accepted).orElseThrow();
+ message.field4 = "changed after admission";
+ message.putAttr("routing-only", "not accepted input");
+ message.tlvs.clear();
+ var codec = new PendingRecordCodec();
+ var record = new PendingSourceRecord(accepted, PendingSourceRecord.Disposition.READY, PendingMessageData.capture(retained));
+ StandardMessage restored = codec.decode(codec.encode(record)).message().restore();
+ assertThat(restored).usingRecursiveComparison().isEqualTo(retained);
+ assertThat(restored.field1).isEqualTo(7);
+ assertThat(restored.field2).isEqualTo(8);
+ assertThat(restored.field3).isEqualTo(1);
+ assertThat(restored.field4).isEqualTo("billing-info");
+ assertThat(restored.dlrReturnMetadata).isEqualTo(metadata);
+ assertThat(restored.tlvs).containsEntry("0427", "1");
+ assertThat(restored.attrs).containsExactlyEntriesOf(Map.of("request", "original"));
+ }
+ }
+ }
+
@Test
void snapshotDetachesProtocolCollectionsAndMutableAttributeValues() {
var message = new StandardMessage();
diff --git a/sendium-core/pom.xml b/sendium-core/pom.xml
index 69fcb659..c4061450 100644
--- a/sendium-core/pom.xml
+++ b/sendium-core/pom.xml
@@ -70,6 +70,10 @@
com.fasterxml.jackson.dataformatjackson-dataformat-yaml
+
+ com.fasterxml.jackson.dataformat
+ jackson-dataformat-cbor
+ io.quarkusquarkus-junit
diff --git a/sendium-core/src/main/java/gr/cytech/sendium/core/http/KannelResource.java b/sendium-core/src/main/java/gr/cytech/sendium/core/http/KannelResource.java
index 258e41a0..67248ca8 100644
--- a/sendium-core/src/main/java/gr/cytech/sendium/core/http/KannelResource.java
+++ b/sendium-core/src/main/java/gr/cytech/sendium/core/http/KannelResource.java
@@ -217,17 +217,20 @@ public Response receiveSms(
if (msg.acked) {
msg.dlrReturnMetadata = DlrReturnMetadata.http(usr, msg.from, msg.to, dlrUrl);
}
+ String acceptedMessageId;
if (outboundCoordinators.isUnsatisfied()) {
queueProvider.getRouterQueue().enqueue(msg);
+ acceptedMessageId = msg.serial;
} else {
- outboundCoordinators.get().accept(new SourceId(gatewayMessageId), msg);
+ SourceId accepted = outboundCoordinators.get().accept(new SourceId(gatewayMessageId), msg);
+ acceptedMessageId = accepted.value().toString();
}
if (MessageTrace.shouldLog(configurationHandler, MessageTrace.EVENT_ACCEPTED)) {
logger.info("message.accepted ingress=http {}", MessageTrace.identifiers(msg));
}
return Response.status(Response.Status.ACCEPTED)
- .entity(msg.serial)
+ .entity(acceptedMessageId)
.build();
} catch (OutboundStorageException e) {
diff --git a/sendium-core/src/main/java/gr/cytech/sendium/core/outbound/DefaultOutboundCoordinator.java b/sendium-core/src/main/java/gr/cytech/sendium/core/outbound/DefaultOutboundCoordinator.java
index 2cc09cf1..d7007c83 100644
--- a/sendium-core/src/main/java/gr/cytech/sendium/core/outbound/DefaultOutboundCoordinator.java
+++ b/sendium-core/src/main/java/gr/cytech/sendium/core/outbound/DefaultOutboundCoordinator.java
@@ -15,6 +15,7 @@
import gr.cytech.sendium.core.storage.PendingMessageStore;
import gr.cytech.sendium.core.storage.RoutedWorkStore;
import gr.cytech.sendium.core.storage.SelectedRouterStore;
+import gr.cytech.sendium.core.storage.pending.MultipartMembership;
import java.time.Duration;
import java.util.ArrayList;
@@ -122,32 +123,58 @@ public void beginShutdown() {
}
@Override
- public void accept(SourceId source, M message) {
- accept(source, message, false);
+ public SourceId accept(SourceId source, M message) {
+ return accept(source, message, false);
}
- private void accept(SourceId source, M message, boolean held) {
+ private SourceId accept(SourceId source, M message, boolean held) {
Objects.requireNonNull(source, "source");
Objects.requireNonNull(message, "message");
lock.lock();
try {
requireRunning(Role.PENDING);
if (sourceSelections.containsKey(source)) {
- return;
+ return source;
}
- if (held) {
- pending.admitHeld(source, message);
- } else {
- pending.admit(source, message);
+ SourceId accepted = held ? pending.admitHeld(source, message) : pending.admit(source, message);
+ if (!source.equals(accepted)) {
+ throw new OutboundStorageException(Role.PENDING, OutboundStorageException.Reason.OWNERSHIP_CONFLICT,
+ "Pending admission did not confirm the requested source handle");
}
+ return accepted;
} finally {
lock.unlock();
}
}
@Override
- public void acceptHeld(SourceId source, M message) {
- accept(source, message, true);
+ public SourceId acceptHeld(SourceId source, M message) {
+ return accept(source, message, true);
+ }
+
+ @Override
+ public MultipartMembership acceptPart(SourceId source, M message, Duration timeout) {
+ Objects.requireNonNull(source, "source");
+ Objects.requireNonNull(message, "message");
+ Objects.requireNonNull(timeout, "timeout");
+ lock.lock();
+ try {
+ requireRunning(Role.PENDING);
+ MultipartMembership accepted;
+ if (sourceSelections.containsKey(source)) {
+ accepted = pending.findMultipart(source)
+ .orElseThrow(() -> invalid(Role.PENDING, "Multipart source is no longer retained"));
+ } else {
+ accepted = pending.admitPart(source, message, timeout);
+ }
+ if (accepted == null || !source.equals(accepted.source())) {
+ throw new OutboundStorageException(Role.PENDING, OutboundStorageException.Reason.OWNERSHIP_CONFLICT,
+ "Pending admission did not confirm multipart source membership");
+ }
+ return accepted;
+ } finally {
+ lock.unlock();
+ }
}
@Override
diff --git a/sendium-core/src/main/java/gr/cytech/sendium/core/outbound/OutboundCoordinator.java b/sendium-core/src/main/java/gr/cytech/sendium/core/outbound/OutboundCoordinator.java
index 35186325..4ba26f35 100644
--- a/sendium-core/src/main/java/gr/cytech/sendium/core/outbound/OutboundCoordinator.java
+++ b/sendium-core/src/main/java/gr/cytech/sendium/core/outbound/OutboundCoordinator.java
@@ -7,6 +7,7 @@
import gr.cytech.sendium.core.outbound.OutboundWork.SelectionId;
import gr.cytech.sendium.core.outbound.OutboundWork.SourceId;
import gr.cytech.sendium.core.outbound.OutboundWork.WorkId;
+import gr.cytech.sendium.core.storage.pending.MultipartMembership;
import java.time.Duration;
import java.util.Optional;
@@ -35,11 +36,14 @@ public interface OutboundCoordinator extends AutoClos
*/
void beginShutdown();
- /** Accepts source ownership and makes the message ready for selection according to the store policy. */
- void accept(SourceId source, M message);
+ /** Accepts ready source ownership and returns the same confirmed handle before protocol success; null is never acceptance. */
+ SourceId accept(SourceId source, M message);
- /** Accepts source ownership without making this source routable yet. */
- void acceptHeld(SourceId source, M message);
+ /** Accepts ownership without making the source routable; returns the same confirmed handle as accept. */
+ SourceId acceptHeld(SourceId source, M message);
+
+ /** Accepts a multipart source with stable group membership before protocol success. */
+ MultipartMembership acceptPart(SourceId source, M message, Duration timeout);
/**
* Makes already-accepted held sources available to bounded selection using one prepared execution message.
diff --git a/sendium-core/src/main/java/gr/cytech/sendium/core/smpp/server/MessagePartsHandler.java b/sendium-core/src/main/java/gr/cytech/sendium/core/smpp/server/MessagePartsHandler.java
index 63622466..bc2fc854 100644
--- a/sendium-core/src/main/java/gr/cytech/sendium/core/smpp/server/MessagePartsHandler.java
+++ b/sendium-core/src/main/java/gr/cytech/sendium/core/smpp/server/MessagePartsHandler.java
@@ -1,10 +1,18 @@
package gr.cytech.sendium.core.smpp.server;
import gr.cytech.sendium.core.message.StandardMessage;
+import gr.cytech.sendium.core.outbound.OutboundWork.SourceId;
+import gr.cytech.sendium.core.storage.PendingMessageStore;
+import gr.cytech.sendium.core.storage.pending.MultipartKey;
+import gr.cytech.sendium.core.storage.pending.MultipartMembership;
+import gr.cytech.sendium.core.storage.pending.PendingMultipartRecovery;
import gr.cytech.sendium.util.MessageUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
@@ -17,6 +25,7 @@
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
/**
* This class implements a handler that collects message parts for various concatenated
@@ -31,8 +40,9 @@ public class MessagePartsHandler {
private final Map> pendingMessageParts;
private final Map> scheduledTasks;
private final MessagePartsEventsListener listener;
+ private PendingMultipartRecovery recovery;
- private long timeoutInMilliseconds;
+ private volatile long timeoutInMilliseconds;
private ScheduledThreadPoolExecutor executor;
public MessagePartsHandler(MessagePartsEventsListener listener, long timeoutInMilliseconds, ScheduledThreadPoolExecutor executor) {
@@ -56,6 +66,9 @@ public MessagePartsHandler(MessagePartsEventsListener listener, long timeoutI
public synchronized boolean stop() {
logger.info("Stopping MessagePartsHandler: {} ...", listener.getName());
+ if (recovery != null) {
+ recovery.close();
+ }
//process any scheduled tasks and stop the executor
executor.shutdown();
@@ -73,6 +86,54 @@ public void setTimeoutInMilliseconds(long timeoutInMilliseconds) {
this.timeoutInMilliseconds = timeoutInMilliseconds;
}
+ public long getTimeoutInMilliseconds() {
+ return timeoutInMilliseconds;
+ }
+
+ /** Rebuild from a quiescent pending store; the caller supplies its existing coordinated publication boundary. */
+ public synchronized void reconstruct(PendingMessageStore pending, int pageSize, BiConsumer, M> publisher) {
+ if (!pendingMessageParts.isEmpty()) {
+ throw new IllegalStateException("Reconstruction must precede runtime part handling");
+ }
+ if (recovery != null) {
+ recovery.close();
+ }
+ recovery = new PendingMultipartRecovery<>(pending, Clock.systemUTC(), executor, publisher);
+ recovery.reconstruct(pageSize);
+ }
+
+ public synchronized boolean isRecordDriven() {
+ return recovery != null;
+ }
+
+ public synchronized boolean isRecoveryReady() {
+ return recovery != null && recovery.isReady();
+ }
+
+ public synchronized void addAcceptedSource(SourceId source) {
+ if (recovery == null) {
+ throw new IllegalStateException("Record-driven reconstruction has not been selected");
+ }
+ recovery.accepted(source);
+ }
+
+ /** Uses the admission-time group instance and deadline instead of reassigning by a legacy reference. */
+ public synchronized void addAcceptedPart(M part, MultipartMembership membership) {
+ if (!isSupportedMessagePart(part) || !membership.source().value().toString().equalsIgnoreCase(part.serial) ||
+ !membership.key().equals(MultipartKey.from(part)) || membership.ordinal() != MessageUtil.getNumberOfCurrentPart(part)) {
+ throw new IllegalArgumentException("Part does not match its accepted membership");
+ }
+ if (recovery != null) {
+ recovery.accepted(membership.source());
+ return;
+ }
+ if (membership.duplicate()) {
+ return;
+ }
+ long remaining = Math.max(0, Duration.between(Instant.now(), membership.deadline()).toMillis());
+ addMessagePart(part, membership.groupId().toString(), remaining);
+ }
+
/**
* Adds the part in the list of the received parts of the provided message. The message reference
* is extracted based on the UDH information in the binheader field. In case the current part
@@ -88,7 +149,10 @@ public synchronized void addMessagePart(M part) throws IllegalArgumentException
throw new IllegalArgumentException("The provided message part does not contain a supported UDH");
}
String msgRefNum = MessageUtil.getMessageReference(part);
+ addMessagePart(part, msgRefNum, timeoutInMilliseconds);
+ }
+ private void addMessagePart(M part, String msgRefNum, long delay) {
//Create an empty message parts container if such not exists
SortedSet receivedParts = pendingMessageParts.computeIfAbsent(msgRefNum,
k -> new ConcurrentSkipListSet<>(new MessagePartsComparator()));
@@ -98,7 +162,7 @@ public synchronized void addMessagePart(M part) throws IllegalArgumentException
listener.onDuplicateMessagePart(receivedParts.tailSet(part).first(), part);
}
- scheduledTasks.computeIfAbsent(msgRefNum, this::scheduleDelayedMessagePartsTask);
+ scheduledTasks.computeIfAbsent(msgRefNum, key -> scheduleDelayedMessagePartsTask(key, delay));
//check if we have received all the message parts
if (receivedParts.size() == MessageUtil.getNumberOfTotalParts(part)) {
@@ -111,10 +175,10 @@ public synchronized void addMessagePart(M part) throws IllegalArgumentException
}
}
- private Future scheduleDelayedMessagePartsTask(String msgRefNum) {
+ private Future scheduleDelayedMessagePartsTask(String msgRefNum, long delay) {
//schedule a task to notify the listener when a timeout occurs
Callable delayedCall = new DelayedMessagePartsTask(msgRefNum);
- Future futureCall = executor.schedule(delayedCall, timeoutInMilliseconds, TimeUnit.MILLISECONDS);
+ Future futureCall = executor.schedule(delayedCall, delay, TimeUnit.MILLISECONDS);
return futureCall;
}
diff --git a/sendium-core/src/main/java/gr/cytech/sendium/core/smpp/server/SmppServerWorker.java b/sendium-core/src/main/java/gr/cytech/sendium/core/smpp/server/SmppServerWorker.java
index 7d10c067..27deddb1 100644
--- a/sendium-core/src/main/java/gr/cytech/sendium/core/smpp/server/SmppServerWorker.java
+++ b/sendium-core/src/main/java/gr/cytech/sendium/core/smpp/server/SmppServerWorker.java
@@ -30,6 +30,9 @@
import gr.cytech.sendium.core.smpp.util.SmppServerUtil;
import gr.cytech.sendium.core.storage.OutboundStage;
import gr.cytech.sendium.core.storage.OutboundStorageException;
+import gr.cytech.sendium.core.storage.PendingMessageStore;
+import gr.cytech.sendium.core.storage.pending.MultipartMembership;
+import gr.cytech.sendium.core.storage.pending.PendingMultipartRecovery;
import gr.cytech.sendium.core.worker.FailDelayPolicy;
import gr.cytech.sendium.core.worker.FailDelayPolicyAction;
import gr.cytech.sendium.core.worker.WorkerType;
@@ -47,6 +50,7 @@
import java.io.IOException;
import java.sql.Timestamp;
+import java.time.Duration;
import java.time.ZoneOffset;
import java.time.ZonedDateTime;
import java.util.ArrayList;
@@ -148,7 +152,7 @@ public class SmppServerWorker extends AbstractOutWork
protected MessagePartsHandler messagePartsHandler;
protected boolean isFastUnsafeStop = false;
private final ReentrantReadWriteLock ingressLifecycleLock = new ReentrantReadWriteLock(true);
- private final Map> pendingPartSources = Collections.synchronizedMap(new IdentityHashMap<>());
+ private final Map pendingPartSources = Collections.synchronizedMap(new IdentityHashMap<>());
private volatile OutboundCoordinator ingressCoordinator;
private volatile boolean ingressStarted;
@@ -221,6 +225,19 @@ public void setIngressCoordinator(OutboundCoordinator coordinator) {
}
}
+ /** Reconstruct accepted work before startup; this neither opens a backend nor enables a durable profile. */
+ public void reconstructPendingParts(PendingMessageStore pending, int pageSize) {
+ ingressLifecycleLock.writeLock().lock();
+ try {
+ if (ingressStarted || inExecutor != null || ingressCoordinator == null) {
+ throw new IllegalStateException("Pending reconstruction requires a selected coordinator before startup/admission");
+ }
+ messagePartsHandler.reconstruct(pending, pageSize, ingressCoordinator::makeHeldReady);
+ } finally {
+ ingressLifecycleLock.writeLock().unlock();
+ }
+ }
+
@Override
public String getType() {
return TYPE_SMPP_SERVER;
@@ -230,6 +247,7 @@ public String getType() {
public Thread start() {
ingressLifecycleLock.writeLock().lock();
try {
+ requirePendingRecoveryReady();
ingressStarted = true;
} finally {
ingressLifecycleLock.writeLock().unlock();
@@ -625,6 +643,10 @@ protected boolean configFlagReverseDlrSrcDst() {
protected void configReassemblingTimeout() {
long timeout = configurationProvider.getLongPrpt(_timeoutMillis);
+ if (messagePartsHandler != null && messagePartsHandler.isRecordDriven()) {
+ messagePartsHandler.setTimeoutInMilliseconds(timeout > 0 ? timeout : 30000L);
+ return;
+ }
if (timeout > 0) {
setMessagePartsHandler(new MessagePartsHandler(new CcatMessagePartsEventsListener(), configurationProvider.getLongPrpt(_timeoutMillis)));
logger.warn("configReassemblingTimeout|warn|set:{}", timeout);
@@ -1018,22 +1040,29 @@ private void acceptIngress(InEvent ine) {
private boolean admitToLifecycle(InEvent event, UUID gatewayMessageId) {
try {
+ requirePendingRecoveryReady();
SourceId source = new SourceId(gatewayMessageId);
if (event.submitSm != null && !Strings.isNullOrEmpty(event.pMsg.binheader)) {
if (!messagePartsHandler.isSupportedMessagePart(event.pMsg)) {
throw new OutboundStorageException(OutboundStage.Role.PENDING, OutboundStorageException.Reason.UNSUPPORTED,
"Unsupported multipart header");
}
- ingressCoordinator.acceptHeld(source, event.pMsg);
- synchronized (pendingPartSources) {
- pendingPartSources.computeIfAbsent(event.pMsg, ignored -> new HashSet<>()).add(source);
+ var membership = ingressCoordinator.acceptPart(source, event.pMsg,
+ Duration.ofMillis(messagePartsHandler.getTimeoutInMilliseconds()));
+ SourceId acceptedSource = membership.source();
+ event.pMsg.serial = acceptedSource.value().toString();
+ if (!messagePartsHandler.isRecordDriven()) {
+ synchronized (pendingPartSources) {
+ pendingPartSources.put(event.pMsg, membership);
+ }
}
InEvent accepted = new InEvent<>(event.pMsg, event.submitSm, event.mpid, event.localTimestamp,
- false, event.responseMessageId, Set.of(source));
+ false, event.responseMessageId, Set.of(acceptedSource));
accepted.notifyClient = event.notifyClient;
inEventQueue.add(accepted);
} else {
- ingressCoordinator.accept(source, event.pMsg);
+ SourceId acceptedSource = ingressCoordinator.accept(source, event.pMsg);
+ event.pMsg.serial = acceptedSource.value().toString();
}
if (MessageTrace.shouldLog(configurationProvider, MessageTrace.EVENT_ACCEPTED)) {
logger.info("message.accepted ingress=smppserver worker={} {}", getFullName(), MessageTrace.identifiers(event.pMsg));
@@ -1062,6 +1091,13 @@ private void rejectIngress(InEvent event, int status) {
}
}
+ private void requirePendingRecoveryReady() {
+ if (messagePartsHandler != null && messagePartsHandler.isRecordDriven() && !messagePartsHandler.isRecoveryReady()) {
+ throw new OutboundStorageException(OutboundStage.Role.PENDING, OutboundStorageException.Reason.UNAVAILABLE,
+ "Pending multipart reconstruction is not ready");
+ }
+ }
+
protected boolean checkReassembling(M msg) {
if (!this.keepOnRunning) {
return true;
@@ -1091,8 +1127,16 @@ public void handleIngressMessages(List> events) {
}
if (event.submitSm == null) {
ingressCoordinator.makeHeldReady(event.sourceIds, event.pMsg);
- } else if (pendingPartSources.containsKey(event.pMsg)) {
- messagePartsHandler.addMessagePart(event.pMsg);
+ } else if (messagePartsHandler.isRecordDriven()) {
+ event.sourceIds.forEach(messagePartsHandler::addAcceptedSource);
+ } else {
+ var membership = pendingPartSources.get(event.pMsg);
+ if (membership != null) {
+ messagePartsHandler.addAcceptedPart(event.pMsg, membership);
+ if (membership.duplicate()) {
+ pendingPartSources.remove(event.pMsg);
+ }
+ }
}
// A replayed raw event may already have handed its sources to a retained part or ready event.
continue;
@@ -1409,19 +1453,6 @@ private int getMessageType(StandardMessage msg) throws IOException {
}
public class CcatMessagePartsEventsListener implements MessagePartsEventsListener {
- @Override
- public void onDuplicateMessagePart(M original, M duplicate) {
- if (ingressCoordinator == null || original == duplicate) {
- return;
- }
- synchronized (pendingPartSources) {
- Set originalSources = requirePartSources(original);
- Set duplicateSources = requirePartSources(duplicate);
- originalSources.addAll(duplicateSources);
- pendingPartSources.remove(duplicate);
- }
- }
-
@Override
public void onMessagePartsHandlingEvent(MessagePartsHandler.MessagePartsEventType type, List parts) {
@@ -1435,11 +1466,7 @@ public void onMessagePartsHandlingEvent(MessagePartsHandler.MessagePartsEventTyp
}
if (type == MessagePartsHandler.MessagePartsEventType.COMPLETE) {
//The first part contains the id that we need to send to the provider
- M message = parts.getFirst();
- //we make sure that the filter will not process again the reassembled message
- message.binheader = null;
- message.body = parts.stream().map(m -> m.body).collect(Collectors.joining(""));
- message.reassembledParts = parts.stream().map(m -> m.serial).collect(Collectors.toCollection(ArrayList::new));
+ M message = PendingMultipartRecovery.assemble(parts);
Set allSources = sources.stream().flatMap(Set::stream).collect(Collectors.toSet());
InEvent event = new InEvent<>(message, null, message.onetwork, new Timestamp(message.ctstamp),
@@ -1462,11 +1489,11 @@ public void onMessagePartsHandlingEvent(MessagePartsHandler.MessagePartsEventTyp
}
private Set requirePartSources(M part) {
- Set sources = pendingPartSources.get(part);
- if (sources == null || sources.isEmpty()) {
+ var membership = pendingPartSources.get(part);
+ if (membership == null) {
throw new IllegalStateException("Reassembled part has no lifecycle source ownership");
}
- return sources;
+ return Set.of(membership.source());
}
@Override
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
index dd15e829..d948f306 100644
--- 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
@@ -2,8 +2,13 @@
import gr.cytech.sendium.core.message.StandardMessage;
import gr.cytech.sendium.core.outbound.OutboundWork.SourceId;
+import gr.cytech.sendium.core.storage.pending.MultipartMembership;
+import java.time.Duration;
+import java.util.List;
+import java.util.Optional;
import java.util.Set;
+import java.util.UUID;
public interface PendingMessageStore extends OutboundStage {
/**
@@ -13,11 +18,46 @@ public interface PendingMessageStore extends Outbound
* 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.
+ * Returns the supplied stable source handle only after backend acceptance; never returns null
+ * or substitutes another source ID. Callers must use this result before acknowledging success.
*/
- void admit(SourceId source, M message);
+ SourceId admit(SourceId source, M message);
/** Retains an accepted source without making it eligible for selection until makeHeldReady succeeds. */
- void admitHeld(SourceId source, M message);
+ SourceId admitHeld(SourceId source, M message);
+
+ /**
+ * Atomically accepts a held part with its immutable membership before acknowledgement. Join the open,
+ * unexpired matching group or create a new group. The first accepted source for an ordinal wins;
+ * later sources report duplicate membership referencing that original source, without replacing its payload.
+ * All accepted handles remain owned until terminal completion. The first part fixes the deadline using timeout;
+ * later arrivals never extend it. An internal retry of a retained multipart source returns its original membership.
+ * A source already admitted without multipart membership cannot be converted through this method.
+ */
+ MultipartMembership admitPart(SourceId source, M message, Duration timeout);
+
+ /** Returns recorded membership for a retained multipart source; ordinary or absent sources have no membership. */
+ Optional findMultipart(SourceId source);
+
+ /**
+ * Returns retained handles assigned to one group, including duplicates, bounded by pending source capacity.
+ * Includes members awaiting ingress processing, so expiry does not depend on notification order.
+ * Unknown or fully completed groups return an empty list; backend failures must still propagate.
+ */
+ List multipartSources(UUID groupId);
+
+ /** Returns an isolated accepted payload, or absence established by a ready backend; failures are not absence. */
+ Optional find(SourceId source);
+
+ /**
+ * Returns at most limit retained source handles, including held/selected/routed sources, in UUID.compareTo
+ * order strictly after afterExclusive. Pass null to start; continue from the last returned ID, even if that
+ * source has since completed. An empty page ends traversal. limit must be positive. Pages are immutable.
+ * This is bounded recovery access, not selection or a payload backlog copy. A quiescent scan visits all
+ * retained sources once. Concurrent admission at/before the cursor may be missed; completed sources may
+ * disappear before find. Startup recovery must quiesce admission/processing rather than assume a snapshot.
+ */
+ List sourcesAfter(SourceId afterExclusive, int limit);
/**
* Makes held sources ready using one execution message referring to their nonempty source set. Original
@@ -26,12 +66,19 @@ public interface PendingMessageStore extends Outbound
* 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.
+ * Assigned multipart ordinals may be published together only as their complete recorded group;
+ * partial/expired ordinals are published individually, never mixed with another group or unassigned held sources.
+ * Publication of original sources atomically includes every retained duplicate of those ordinals, even if
+ * its ingress notification has not been processed. A duplicate cannot be published without its original.
*/
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.
+ * Called only after all work and required handoffs are terminal. Successful return confirms that
+ * every requested source is absent. Missing sources count as complete only when a ready backend can
+ * establish their absence; unavailability, corrupt state, or inaccessible ownership must fail explicitly.
+ * A failed call may have removed some or all requested sources before its result became uncertain.
+ * Retrying the full source set must therefore be safe; a failure must never be treated as established absence.
*/
void complete(Set sources);
}
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
index f9e2abfb..4baa8d39 100644
--- 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
@@ -4,10 +4,18 @@
import gr.cytech.sendium.core.outbound.OutboundWork.SourceId;
import gr.cytech.sendium.core.storage.OutboundStorageException;
import gr.cytech.sendium.core.storage.PendingMessageStore;
+import gr.cytech.sendium.core.storage.pending.MultipartKey;
+import gr.cytech.sendium.core.storage.pending.MultipartMembership;
+import gr.cytech.sendium.util.MessageUtil;
+import java.time.Clock;
+import java.time.Duration;
import java.time.Instant;
+import java.util.ArrayList;
+import java.util.BitSet;
import java.util.Comparator;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.NavigableSet;
@@ -16,6 +24,7 @@
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
+import java.util.UUID;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
@@ -39,11 +48,15 @@ public final class MemoryPendingMessageStore implemen
private final int capacity;
private final UnaryOperator snapshots;
private final Function super M, Instant> eligibleAt;
- private final Map> records = new HashMap<>();
+ private final Clock clock;
+ private final NavigableMap> records = new TreeMap<>(Comparator.comparing(SourceId::value));
+ private final Map> openGroups = new HashMap<>();
+ private final Map retainedGroups = new HashMap<>();
private final NavigableMap>> waiting = new TreeMap<>(Comparator.reverseOrder());
private State state = State.NEW;
private Object selectorOwner;
private long sequence;
+ private long multipartSequence;
public MemoryPendingMessageStore(int capacity, UnaryOperator snapshots) {
this(capacity, snapshots, message -> Instant.MIN);
@@ -51,12 +64,17 @@ public MemoryPendingMessageStore(int capacity, UnaryOperator snapshots) {
/** Eligibility is evaluated when ready work is first published; held admission does not evaluate it. */
public MemoryPendingMessageStore(int capacity, UnaryOperator snapshots, Function super M, Instant> eligibleAt) {
+ this(capacity, snapshots, eligibleAt, Clock.systemUTC());
+ }
+
+ public MemoryPendingMessageStore(int capacity, UnaryOperator snapshots, Function super M, Instant> eligibleAt, Clock clock) {
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");
+ this.clock = Objects.requireNonNull(clock, "clock");
}
@Override
@@ -83,18 +101,18 @@ public Status status() {
}
@Override
- public void admit(SourceId source, M message) {
- admit(source, message, false);
+ public SourceId admit(SourceId source, M message) {
+ return admit(source, message, false);
}
- private void admit(SourceId source, M message, boolean held) {
+ private SourceId 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;
+ return source;
}
if (records.size() >= capacity) {
throw new OutboundStorageException(Role.PENDING, CAPACITY_EXCEEDED, "Memory pending capacity reached");
@@ -107,19 +125,121 @@ private void admit(SourceId source, M message, boolean held) {
accepted.ready = ready;
addWaiting(ready);
}
+ return source;
} finally {
lock.unlock();
}
}
@Override
- public void admitHeld(SourceId source, M message) {
- admit(source, message, true);
+ public SourceId admitHeld(SourceId source, M message) {
+ return admit(source, message, true);
+ }
+
+ @Override
+ public MultipartMembership admitPart(SourceId source, M message, Duration timeout) {
+ Objects.requireNonNull(source, "source");
+ Objects.requireNonNull(message, "message");
+ Objects.requireNonNull(timeout, "timeout");
+ if (timeout.isZero() || timeout.isNegative()) {
+ throw new IllegalArgumentException("Multipart timeout must be positive");
+ }
+ lock.lock();
+ try {
+ requireReady(Role.PENDING);
+ Source existing = records.get(source);
+ if (existing != null) {
+ if (existing.membership == null) {
+ throw new OutboundStorageException(Role.PENDING, INVALID_TRANSITION, "Source has no multipart membership");
+ }
+ return existing.membership;
+ }
+ if (records.size() >= capacity) {
+ throw new OutboundStorageException(Role.PENDING, CAPACITY_EXCEEDED, "Memory pending capacity reached");
+ }
+ M snapshot = snapshot(message, Role.PENDING);
+ MultipartKey key;
+ int ordinal;
+ try {
+ if (!source.value().toString().equalsIgnoreCase(snapshot.serial)) {
+ throw new IllegalArgumentException("Multipart source identity mismatch");
+ }
+ key = MultipartKey.from(snapshot);
+ ordinal = MessageUtil.concatenation(snapshot).orElseThrow().ordinal();
+ } catch (IllegalArgumentException failure) {
+ throw new OutboundStorageException(Role.PENDING, UNSUPPORTED, "Unsupported multipart identity");
+ }
+ Instant now = clock.instant();
+ var groups = openGroups.getOrDefault(key, new ArrayList<>());
+ groups.removeIf(group -> !group.deadline.isAfter(now));
+ MultipartGroup chosen = null;
+ for (MultipartGroup group : groups) {
+ chosen = group;
+ break;
+ }
+ if (chosen == null) {
+ chosen = new MultipartGroup(key, now.plus(timeout), multipartSequence++);
+ groups.add(chosen);
+ }
+ SourceId original = chosen.sources.getOrDefault(ordinal, source);
+ var membership = new MultipartMembership(source, chosen.id, key, ordinal, chosen.deadline, chosen.order, original);
+ records.put(source, new Source<>(snapshot, membership, chosen));
+ chosen.sources.putIfAbsent(ordinal, source);
+ chosen.allSources.add(source);
+ retainedGroups.put(chosen.id, chosen);
+ openGroups.put(key, groups);
+ chosen.ordinals.set(ordinal);
+ if (chosen.ordinals.cardinality() == key.total()) {
+ closeGroup(chosen);
+ }
+ return membership;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public Optional findMultipart(SourceId source) {
+ Objects.requireNonNull(source, "source");
+ lock.lock();
+ try {
+ requireReady(Role.PENDING);
+ Source accepted = records.get(source);
+ return accepted == null ? Optional.empty() : Optional.ofNullable(accepted.membership);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private void closeGroup(MultipartGroup group) {
+ if (group == null) {
+ return;
+ }
+ List groups = openGroups.get(group.key);
+ if (groups != null) {
+ groups.remove(group);
+ if (groups.isEmpty()) {
+ openGroups.remove(group.key);
+ }
+ }
+ }
+
+ @Override
+ public List multipartSources(UUID groupId) {
+ Objects.requireNonNull(groupId, "groupId");
+ lock.lock();
+ try {
+ requireReady(Role.PENDING);
+ MultipartGroup group = retainedGroups.get(groupId);
+ return group == null ? List.of() : List.copyOf(group.allSources);
+ } finally {
+ lock.unlock();
+ }
}
@Override
public void makeHeldReady(Set sources, M message) {
- Set checked = Set.copyOf(sources);
+ Set checked = new java.util.HashSet<>(sources);
Objects.requireNonNull(message, "message");
if (checked.isEmpty()) {
throw new IllegalArgumentException("Ready work must reference at least one source");
@@ -128,11 +248,18 @@ public void makeHeldReady(Set sources, M message) {
try {
requireReady(Role.PENDING);
Entry existing = null;
+ MultipartGroup assignedGroup = null;
for (SourceId source : checked) {
Source accepted = records.get(source);
if (accepted == null || !accepted.initiallyHeld) {
throw invalidPublication();
}
+ if (accepted.group != null) {
+ if (assignedGroup != null && assignedGroup != accepted.group) {
+ throw invalidPublication();
+ }
+ assignedGroup = accepted.group;
+ }
if (accepted.ready != null) {
if (existing != null && accepted.ready != existing) {
throw invalidPublication();
@@ -140,6 +267,24 @@ public void makeHeldReady(Set sources, M message) {
existing = accepted.ready;
}
}
+ if (assignedGroup != null) {
+ for (SourceId source : checked) {
+ if (records.get(source).group != assignedGroup || !checked.contains(records.get(source).membership.originalSource())) {
+ throw invalidPublication();
+ }
+ }
+ Set originals = checked.stream().map(source -> records.get(source).membership.originalSource())
+ .collect(java.util.stream.Collectors.toSet());
+ if (originals.size() > 1 && originals.size() != assignedGroup.key.total()) {
+ throw invalidPublication();
+ }
+ for (SourceId source : assignedGroup.allSources) {
+ if (originals.contains(records.get(source).membership.originalSource())) {
+ checked.add(source);
+ }
+ }
+ }
+ checked = Set.copyOf(checked);
if (existing != null) {
if (!existing.sources.equals(checked)) {
throw invalidPublication();
@@ -154,6 +299,7 @@ public void makeHeldReady(Set sources, M message) {
Entry entry = readyEntry(checked, snapshot(message, Role.PENDING));
for (SourceId source : checked) {
records.get(source).ready = entry;
+ closeGroup(records.get(source).group);
}
addWaiting(entry);
} finally {
@@ -182,7 +328,7 @@ private static OutboundStorageException invalidPublication() {
"Ready publication requires retained held sources with one matching source set");
}
- /** Single-source lookup for memory-backend consumers; returns an isolated execution value. */
+ @Override
public Optional find(SourceId source) {
Objects.requireNonNull(source, "source");
lock.lock();
@@ -195,6 +341,28 @@ public Optional find(SourceId source) {
}
}
+ @Override
+ public List sourcesAfter(SourceId afterExclusive, int limit) {
+ if (limit <= 0) {
+ throw new IllegalArgumentException("Recovery page limit must be positive");
+ }
+ lock.lock();
+ try {
+ requireReady(Role.PENDING);
+ var remaining = afterExclusive == null ? records : records.tailMap(afterExclusive, false);
+ var page = new ArrayList();
+ for (SourceId source : remaining.navigableKeySet()) {
+ if (page.size() == limit) {
+ break;
+ }
+ page.add(source);
+ }
+ return List.copyOf(page);
+ } finally {
+ lock.unlock();
+ }
+ }
+
@Override
public void complete(Set sources) {
Set checked = Set.copyOf(sources);
@@ -203,8 +371,17 @@ public void complete(Set sources) {
requireReady(Role.PENDING);
for (SourceId source : checked) {
Source entry = records.remove(source);
- if (entry != null && entry.ready != null) {
- removeWaiting(entry.ready);
+ if (entry != null) {
+ closeGroup(entry.group);
+ if (entry.group != null) {
+ entry.group.allSources.remove(source);
+ if (entry.group.allSources.isEmpty()) {
+ retainedGroups.remove(entry.group.id);
+ }
+ }
+ if (entry.ready != null) {
+ removeWaiting(entry.ready);
+ }
}
}
} finally {
@@ -218,6 +395,8 @@ public void close() {
try {
state = State.CLOSED;
records.clear();
+ openGroups.clear();
+ retainedGroups.clear();
waiting.clear();
changed.signalAll();
} finally {
@@ -297,11 +476,38 @@ static final class Entry {
private static final class Source {
private final M message;
private final boolean initiallyHeld;
+ private final MultipartMembership membership;
+ private final MultipartGroup group;
private Entry ready;
private Source(M message, boolean held) {
this.message = message;
this.initiallyHeld = held;
+ this.membership = null;
+ this.group = null;
+ }
+
+ private Source(M message, MultipartMembership membership, MultipartGroup group) {
+ this.message = message;
+ this.initiallyHeld = true;
+ this.membership = membership;
+ this.group = group;
+ }
+ }
+
+ private static final class MultipartGroup {
+ private final UUID id = UUID.randomUUID();
+ private final MultipartKey key;
+ private final Instant deadline;
+ private final long order;
+ private final BitSet ordinals = new BitSet();
+ private final NavigableMap sources = new TreeMap<>();
+ private final Set allSources = new java.util.LinkedHashSet<>();
+
+ private MultipartGroup(MultipartKey key, Instant deadline, long order) {
+ this.key = key;
+ this.deadline = deadline;
+ this.order = order;
}
}
}
diff --git a/sendium-core/src/main/java/gr/cytech/sendium/core/storage/pending/MultipartKey.java b/sendium-core/src/main/java/gr/cytech/sendium/core/storage/pending/MultipartKey.java
new file mode 100644
index 00000000..e0a9d29c
--- /dev/null
+++ b/sendium-core/src/main/java/gr/cytech/sendium/core/storage/pending/MultipartKey.java
@@ -0,0 +1,24 @@
+package gr.cytech.sendium.core.storage.pending;
+
+import gr.cytech.sendium.core.message.StandardMessage;
+import gr.cytech.sendium.util.MessageUtil;
+
+import java.util.Objects;
+
+/** Logical matching identity; several independent group instances may share this key. */
+public record MultipartKey(String accountId, String systemId, String sender, String recipient,
+ int referenceBits, int reference, int total, String normalizedUdh) {
+ public MultipartKey {
+ Objects.requireNonNull(normalizedUdh, "normalizedUdh");
+ if ((referenceBits != 8 && referenceBits != 16) || reference < 0 ||
+ reference > (referenceBits == 8 ? 255 : 65535) || total < 1 || total > 255) {
+ throw new IllegalArgumentException("Invalid multipart identity");
+ }
+ }
+
+ public static MultipartKey from(StandardMessage message) {
+ var part = MessageUtil.concatenation(message).orElseThrow(() -> new IllegalArgumentException("Unsupported multipart header"));
+ return new MultipartKey(message.owner_id, message.systemId, message.from, message.to,
+ part.referenceBits(), part.reference(), part.total(), part.normalizedUdh());
+ }
+}
diff --git a/sendium-core/src/main/java/gr/cytech/sendium/core/storage/pending/MultipartMembership.java b/sendium-core/src/main/java/gr/cytech/sendium/core/storage/pending/MultipartMembership.java
new file mode 100644
index 00000000..2ea4ab8d
--- /dev/null
+++ b/sendium-core/src/main/java/gr/cytech/sendium/core/storage/pending/MultipartMembership.java
@@ -0,0 +1,30 @@
+package gr.cytech.sendium.core.storage.pending;
+
+import gr.cytech.sendium.core.outbound.OutboundWork.SourceId;
+
+import java.time.Instant;
+import java.util.Objects;
+import java.util.UUID;
+
+/** Immutable admission facts: original source UUID, assigned group instance, ordinal and saved deadline. */
+public record MultipartMembership(SourceId source, UUID groupId, MultipartKey key, int ordinal, Instant deadline, long groupOrder,
+ SourceId originalSource) {
+ public MultipartMembership(SourceId source, UUID groupId, MultipartKey key, int ordinal, Instant deadline, long groupOrder) {
+ this(source, groupId, key, ordinal, deadline, groupOrder, source);
+ }
+
+ public MultipartMembership {
+ Objects.requireNonNull(source, "source");
+ Objects.requireNonNull(originalSource, "originalSource");
+ Objects.requireNonNull(groupId, "groupId");
+ Objects.requireNonNull(key, "key");
+ Objects.requireNonNull(deadline, "deadline");
+ if (ordinal < 1 || ordinal > key.total() || groupOrder < 0) {
+ throw new IllegalArgumentException("Invalid multipart membership");
+ }
+ }
+
+ public boolean duplicate() {
+ return !source.equals(originalSource);
+ }
+}
diff --git a/sendium-core/src/main/java/gr/cytech/sendium/core/storage/pending/PendingExtensionValues.java b/sendium-core/src/main/java/gr/cytech/sendium/core/storage/pending/PendingExtensionValues.java
new file mode 100644
index 00000000..b0bb7c5d
--- /dev/null
+++ b/sendium-core/src/main/java/gr/cytech/sendium/core/storage/pending/PendingExtensionValues.java
@@ -0,0 +1,164 @@
+package gr.cytech.sendium.core.storage.pending;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+
+/** A closed value vocabulary; type tags are schema names, never Java class names. */
+final class PendingExtensionValues {
+ private static final JsonNodeFactory NODES = JsonNodeFactory.instance;
+ private static final int MAX_DEPTH = 64;
+
+ private PendingExtensionValues() {
+ }
+
+ static JsonNode encode(Object value) {
+ return encode(value, new IdentityHashMap<>(), 0);
+ }
+
+ private static JsonNode encode(Object value, IdentityHashMap