create a d14n-ordered stream

[insipx]

closes #2790

[insipx]

This PR is part of a stack of 12 bookmarks:

1. trunk()
2. insipx/causal-fix
3. insipx/icebox-global-cursor
4. insipx/batch-queries
5. insipx/extractors
6. insipx/cursor-store-icebox
7. insipx/icebox-in-ordering
8. insipx/test-ordering
9. insipx/paged-stream
10. insipx/ordered-stream ← this PR
11. insipx/ordered-stream-test
12. push-pnvnmqsyqqqy
13. push-syomxrnuqsnk

---

Created with jj-stack

[claude]

Claude finished @insipx's task —— View job

---

Code Review Complete

Review Date: December 15, 2025 20:51 UTC

I've reviewed the d14n-ordered stream implementation. The overall design is solid, introducing offline ordering with orphan icing. Most issues have been identified by macroscopeapp. Here are additional high-priority concerns not already covered:

Critical Issues

1. Infinite Loop Risk in resolve_children_inner (Already noted by macroscopeapp)
xmtp_api_d14n/src/protocol/in_memory_cursor_store.rs:138-165

The recursive implementation lacks cycle detection. While macroscopeapp flagged this, I want to emphasize the severity: if two envelopes form a dependency cycle (A depends on B, B depends on A), this will recurse indefinitely until stack overflow.

2. Missing Test Coverage for OrderedStream
xmtp_api_d14n/src/queries/stream/ordered.rs:74-104

The new OrderedStream has zero direct unit tests. Codecov shows 82.92% coverage, meaning 7 lines are untested. Critical paths missing coverage:

• Error propagation from OrderedStreamError (lines 34-50)
• Stream termination with orphaned envelopes
• Multiple batches with mixed resolvable/unresolvable dependencies

Recommendation: Add tests for:

#[xmtp_common::test]
async fn test_ordered_stream_with_missing_dependencies()
#[xmtp_common::test]  
async fn test_ordered_stream_error_propagation()

High-Priority Issues

3. Incorrect is_retryable() Implementation (Macroscopeapp's latest finding is correct)
xmtp_api_d14n/src/queries/stream/ordered.rs:40-44

impl RetryableError for OrderedStreamError {
    fn is_retryable(&self) -> bool {
        false  // ❌ Always returns false
    }
}

This breaks retry logic for transient ResolutionErrors. Macroscopeapp's suggested fix is correct:

matches!(self, OrderedStreamError::Resolver(err) if err.is_retryable())

4. Error Type Semantics
xmtp_api_d14n/src/queries/stream/ordered.rs:26-32

The comment says "this is an error which should never occur, and if it does is a bug in libxmtp", but the code treats it as a regular error. If this truly represents an invariant violation:

• Consider using panic! or unreachable! for unrecoverable cases
• Or update the comment to reflect actual error semantics (transient ordering failures)

Performance Considerations

5. Cloning TopicCursor on Every Poll
xmtp_api_d14n/src/queries/stream/ordered.rs:95

.topic_cursor(self.topic_cursor.clone())  // HashMap clone on every stream item

TopicCursor is a HashMap<Topic, GlobalCursor>. Cloning on every stream poll could be expensive with many topics. Consider:

• Using Arc<RwLock<TopicCursor>> if mutations are rare
• Or passing by reference and mutating in-place (requires API redesign)

6. Memory Swap for Cursor Update
xmtp_api_d14n/src/queries/stream/ordered.rs:99-101

let (envelopes, mut new_cursor) = ordering.into_parts();
let this = self.as_mut().project();
std::mem::swap(this.topic_cursor, &mut new_cursor);

This is correct but slightly awkward. Direct assignment would be clearer:

*this.topic_cursor = new_cursor;

Code Quality

7. TypedNoopResolver Documentation
xmtp_api_d14n/src/protocol/traits/dependency_resolution.rs:69-97

The new TypedNoopResolver<T> lacks documentation. Add:

/// A typed no-op resolver that returns an empty resolution result.
/// Used in streaming contexts where dependencies are expected to arrive
/// in the stream itself rather than being actively fetched.

Security

No security concerns identified. The implementation correctly:

• Validates dependencies before processing
• Handles malformed envelopes via EnvelopeError
• Ices orphans rather than dropping them

Test Coverage Summary

From codecov report:

• ordered.rs: 82.92% (7 lines missing) - needs stream-level tests
• dependency_resolution.rs: 71.42% (2 lines missing) - TypedNoopResolver partially tested
• traits.rs: 0% (3 lines missing) - likely just type definitions

Recommendations Priority

1. MUST FIX: Cycle detection in resolve_children_inner (macroscopeapp's finding)
2. MUST FIX: is_retryable() implementation (macroscopeapp's finding)
3. SHOULD ADD: Direct unit tests for OrderedStream
4. CONSIDER: Performance optimization for cursor cloning in hot path

---

Overall this is a well-structured implementation that fits cleanly into the existing ordering architecture. The main gaps are in test coverage and the two critical issues flagged by macroscopeapp.

[macroscopeapp]

Add d14n-ordered group message streams and offline causal ordering with icing in queries::d14n::XmtpMlsStreams

Introduce FlattenedStream and OrderedStream adapters and apply them to group message subscriptions to run Ordered::order_offline with a CursorStore, switch the icebox to a HashSet with recursive orphan resolution, and add a typed no-op resolver. Key changes are in streams.rs, ordered.rs, and in_memory_cursor_store.rs.

📍Where to Start

Start with the stream construction and ordering in XmtpMlsStreams at streams.rs, then review the ordering adapter in ordered.rs and the icebox behavior in in_memory_cursor_store.rs.

---

Macroscope summarized 76e0de7.