flattened stream type

[insipx]

related to #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 ← this PR
10. insipx/ordered-stream
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

Reviewed on: December 15, 2025 at 20:17 UTC

Overall, this is a well-implemented stream utility with good test coverage. The code is clean and follows Rust best practices. Here are a few observations:

Code Quality ✓

• Clean abstraction using pin_project_lite for proper pinning
• Good use of existing traits (TryStream, Paged)
• Proper error propagation with generic S::Error: From<EnvelopeError>

Minor Issues

1. Typo in documentation (xmtp_api_d14n/src/queries/stream/flattened.rs:16)

   • "whos" should be "whose" (appears twice: lines 16 and 21)

2. Documentation clarity (xmtp_api_d14n/src/queries/stream/flattened.rs:1)

   • Module-level doc says "whos items" - should be "whose items"

3. Test coverage gap (xmtp_api_d14n/src/queries/stream/flattened.rs:82-128)

   • The test only covers the happy path. Consider adding:
     • A test case with error propagation to verify S::Error handling
     • A test with an empty stream
     • A test with empty Vec items from Paged::messages()

Performance ✓

No concerns - the implementation efficiently wraps the inner stream with minimal overhead using map_ok.

Security ✓

No concerns - straightforward stream transformation with no unsafe code.

---

[macroscopeapp]

Add flattened stream adapter and de-duplicate orphan handling by switching InMemoryCursorStore.icebox to HashSet

Introduce FlattenedStream with flattened() to yield paged message batches, change InMemoryCursorStore.icebox to HashSet<OrphanedEnvelope> with recursive child resolution, and define stable Eq/Hash for OrphanedEnvelope that ignore payload bytes. Update ordering tests and minor display/iterator utilities. See xmtp_api_d14n/src/queries/stream/flattened.rs and xmtp_api_d14n/src/protocol/in_memory_cursor_store.rs.

📍Where to Start

Start with the Stream impl for FlattenedStream in xmtp_api_d14n/src/queries/stream/flattened.rs, then review InMemoryCursorStore changes and resolve_children_inner in xmtp_api_d14n/src/protocol/in_memory_cursor_store.rs.

---

Macroscope summarized 1340914.

[codecov]

Codecov Report

❌ Patch coverage is 93.93939% with 14 lines in your changes missing coverage. Please review.
✅ Project coverage is 73.77%. Comparing base (7f5684f) to head (1340914).
⚠️ Report is 2 commits behind head on main.

Files with missing lines	Patch %	Lines
..._api_d14n/src/protocol/utils/test/test_envelope.rs	76.47%	8 Missing ⚠️
xmtp_api_d14n/src/protocol/order.rs	96.96%	3 Missing ⚠️
xmtp_api_d14n/src/queries/stream/flattened.rs	93.47%	3 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2931      +/-   ##
==========================================
+ Coverage   73.59%   73.77%   +0.18%     
==========================================
  Files         405      406       +1     
  Lines       51710    51929     +219     
==========================================
+ Hits        38057    38312     +255     
+ Misses      13653    13617      -36     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.