refactor(store, fjall, framework)!: stream + codec collapse + wire alignment — PR2 of bytes-envelope refactor

[joeldsouzax]

Summary

PR2 of the bytes-envelope refactor. Builds on #182 (PR1, owned-Bytes envelopes).

This PR does three structurally connected things in one merge window because they touch overlapping crate surface (nexus-store::codec + nexus-store::envelope + nexus-store::stream + nexus-fjall + nexus-store::testing + nexus-framework):

Stream trait collapse

~1,370 lines of GAT-lending stream machinery deleted (cursor.rs, combinators.rs, progress.rs, owned.rs, mod.rs). Replaced with a 50-line marker trait in stream.rs:

pub trait EventStream:
    futures::Stream<Item = Result<PersistedEnvelope, Self::Error>> + Send
{
    type Error: std::error::Error + Send + Sync + 'static;
}

Auto-impl'd for every matching futures::Stream. Combinators come from futures::StreamExt / TryStreamExt for free.

• RawEventStore::Stream<'a> GAT → concrete owned Stream associated type.
• M = () generic dropped from RawEventStore, Subscription, SubscriptionBackend (it was always () in practice).
• FjallStream and FjallSubscriptionStream rewritten as impl futures::Stream with unfold for the subscription's notify/refill cycle. Same for InMemoryStream and InMemorySubscriptionStream.

Codec trait collapse

Decode<E> + BorrowingDecode<E> merged into one Decode<E> with a type Output<'a> GAT:

pub trait Decode<E: ?Sized>: Send + Sync + 'static {
    type Output<'a> where Self: 'a;
    type Error: std::error::Error + Send + Sync + 'static;
    fn decode<'a>(&'a self, env: &'a PersistedEnvelope) -> Result<Self::Output<'a>, Self::Error>;
}

• Owning codecs: Output<'a> = E.
• Borrowing codecs (rkyv): Output<'a> = &'a <E as Archive>::Archived.
• Plain-old-data codecs (bytemuck): Output<'a> = &'a E.

Encode::encode now returns bytes::Bytes instead of Vec<u8> for end-to-end zero-copy from codec to wire.

Wire-format payload-alignment invariant

nexus-store::wire::build_row (added in commit 31a5c34 of this branch) pads every row's payload to a 16-byte boundary inside an AVec-backed Bytes. Both nexus-fjall::FjallStore::append and nexus-store::testing::InMemoryStore::append go through the same builder.

This unblocks BytemuckCodec (zero-copy align_to::<E>) and RkyvCodec (zero-copy access::<E::Archived>) — both ship in this PR as nexus-store feature flags.

Two new codecs

• BytemuckCodec (feature bytemuck) — zero-copy for POD types. Output<'a> = &'a E borrowing from env.payload(). Local BytemuckError wrapper because upstream PodCastError is no_std and doesn't implement std::error::Error. Round-trip test asserts pointer equality between env.payload() and the decoded &E.
• RkyvCodec (feature rkyv) — zero-copy via rkyv 0.8. Output<'a> = &'a <E as Archive>::Archived. Encode via to_bytes_in(value, AlignedVec::new()); decode via access. AlignedVec → Vec<u8> → Bytes loses alignment, but wire::build_row re-aligns on write so the envelope's payload regains it.

Migration

• BorrowingDecode<E> → Decode<E> with type Output<'a> = &'a E. Mechanical for codecs that match one of the two shapes; coherence-conflict only fires if a single codec previously implemented both Decode and BorrowingDecode for the same E (rare).
• Decode::decode(name, payload) → Decode::decode(env) — codecs reach for env.event_type() and env.payload() themselves.
• Encode::encode return type Vec<u8> → Bytes (mechanical via Bytes::from(vec) — zero-copy ownership transfer).
• EventStreamExt::try_fold → futures::TryStreamExt::try_fold (same shape; closure body must now be async move).
• BaseEventStream::to_envelope indirection gone — cursors yield envelopes directly.

Two new StoreError variants

• StoreError::Wire(WireError) — reachable only from upcaster-driven load_with paths where a fresh aligned envelope is synthesized from the transformed morsel via wire::build_row. Practically unreachable for in-budget inputs.
• CodecSnapshotStoreError::Wire(WireError) — the same for snapshot decode (wraps raw snapshot bytes in a synthetic envelope to call the new Decode::decode(env) signature).

Helper

PersistedEnvelope::for_decode(event_type, payload) -> Result<Self, WireError> wraps raw bytes in a synthetic 16-byte-aligned envelope for callers without a real one: upcaster post-transform, snapshot decode, codec round-trip tests.

Test plan

• nix flake check — clippy strict, fmt, taplo, audit, deny, hakari, nextest. Pre-commit hook enforced on every commit in this branch.
• Four wire_alignment_tests.rs cross-cutting tests (sequence, lifecycle, defensive boundary, linearizability) assert payload pointer % 16 == 0 across every yielded envelope.
• wire::tests proptests verify payload alignment across arbitrary inputs.
• Re-enabled zero_copy_save_and_load_roundtrip + zero_copy_multi_save_load PASS (alignment regression from PR1 is now fixed).
• BytemuckCodec round-trip + pointer-equality zero-copy assertion + wrong-size rejection.
• RkyvCodec round-trip on archived fields.
• nexus-fjall::FjallStore conformance via nexus-store-testing harness.
• All adapter conformance, subscription, repository QA, and adversarial property tests pass against the new stream + codec shapes.

Companion docs

• docs/plans/2026-05-27-bytes-envelope-design.md
• docs/plans/2026-05-28-bytes-envelope-pr2-implementation.md
• docs/plans/2026-05-27-bytes-envelope-deviations.md — every divergence from the plan logged with reason + impact, including the Task 6-11 bundling rationale.

Why now (driving trigger)

Postgres adapter realism. sqlx's BYTEA decode path has no Bytes impl and the row buffer's Bytes field is pub(crate) — a borrowed-row-buffer PersistedEnvelope<'a> can never serve a sqlx-style driver cleanly. Once postgres is on the roadmap, GAT lending streams structurally cannot serve all targets. The bytes-envelope pattern unifies fjall (zero-copy under bytes_1), postgres (unavoidable one alloc on decode), and any future driver under one envelope shape.

🤖 Generated with Claude Code