7ef14c2ea9
OpenPay (1 crate, 2174 LOC, 14 tests)
op-payouts Multi-rail payout engine. Visa Direct, Mastercard Send,
ACH credits (NACHA), Fedwire + SWIFT MT103 + ISO 20022
pacs.008, SEPA SCT + SCT Inst (mod-97 IBAN), UK FPS,
FedNow, RTP, PayPal, Wise. Crypto: USDC/USDT on
ETH/Polygon/Base/Solana + BTC + Lightning with real
ERC-20 transfer(address,uint256) calldata. Offline-pure
drivers (emit wire payloads — operators transmit).
Smart Byte (1 crate, 2605 LOC, 38 tests)
smart-byte-nostr Nostr adapter. NIP-01 canonical event id + BIP-340
Schnorr sign/verify; from-scratch bech32 + NIP-19 TLV
(npub/nsec/note/nprofile/nevent/nrelay); NIP-04 legacy
AES-256-CBC DMs; NIP-44 v2 HKDF-SHA256→ChaCha20+HMAC
with bucketed padding; NIP-17 gift-wrap (kind 1059) +
seal (kind 13) + rumor (kind 14); NIP-05/11/13/26/65.
Relay client message vocabulary (REQ/EVENT/CLOSE/AUTH/
EOSE/OK/NOTICE/CLOSED) + in-memory router.
EOC (1 crate, 2024 LOC, 30 tests)
eoc-streaming Token-by-token + tool-call streaming primitives.
SSE framing (data: lines); Anthropic event mapper
(message_start/content_block_*/message_delta/stop/ping);
OpenAI delta mapper; Vercel AI SDK mapper. Bounded-channel
backpressure with watermarks; CancelToken propagation;
Last-Event-ID resumable streams; per-stream joule meter.
Co-Authored-By: Claude Opus 4.7 <[email protected]>
77 lines
2.1 KiB
Rust
77 lines
2.1 KiB
Rust
//! Backpressure: producer awaits when the channel is full and the
|
|
//! receiver is slow.
|
|
|
|
use std::time::Duration;
|
|
|
|
use eoc_streaming::backpressure::{BoundedStream, Watermarks};
|
|
use eoc_streaming::stream::{Event, Role};
|
|
|
|
fn ev() -> Event {
|
|
Event::MessageStart {
|
|
id: None,
|
|
role: Role::Assistant,
|
|
}
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread", start_paused = true)]
|
|
async fn try_send_returns_backpressure_when_full() {
|
|
let mut s = BoundedStream::new(Watermarks {
|
|
capacity: 2,
|
|
high: 2,
|
|
low: 1,
|
|
});
|
|
let p = s.producer();
|
|
p.try_send(ev()).unwrap();
|
|
p.try_send(ev()).unwrap();
|
|
let err = p.try_send(ev()).unwrap_err();
|
|
assert!(matches!(
|
|
err,
|
|
eoc_streaming::error::StreamError::Backpressure
|
|
));
|
|
// Drain one slot; should accept again.
|
|
let _ = s.recv().await;
|
|
p.try_send(ev()).unwrap();
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread", start_paused = true)]
|
|
async fn awaiting_send_pauses_until_receiver_drains() {
|
|
let mut s = BoundedStream::new(Watermarks {
|
|
capacity: 1,
|
|
high: 1,
|
|
low: 0,
|
|
});
|
|
let p = s.producer();
|
|
p.send(ev()).await.unwrap();
|
|
// Now full; spawn a send that should park until we recv.
|
|
let p2 = p.clone();
|
|
let h = tokio::spawn(async move {
|
|
p2.send(ev()).await.unwrap();
|
|
true
|
|
});
|
|
// Time has not advanced — task should still be pending.
|
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
|
assert!(!h.is_finished(), "producer should be parked");
|
|
// Drain one slot — task should now complete.
|
|
let _ = s.recv().await;
|
|
assert!(h.await.unwrap());
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread", start_paused = true)]
|
|
async fn watermark_signals_pause_and_resume() {
|
|
let wm = Watermarks {
|
|
capacity: 4,
|
|
high: 3,
|
|
low: 1,
|
|
};
|
|
let mut s = BoundedStream::new(wm);
|
|
let p = s.producer();
|
|
p.send(ev()).await.unwrap();
|
|
assert!(!p.should_pause());
|
|
p.send(ev()).await.unwrap();
|
|
p.send(ev()).await.unwrap();
|
|
assert!(p.should_pause());
|
|
for _ in 0..3 {
|
|
let _ = s.recv().await;
|
|
}
|
|
assert!(p.should_resume());
|
|
}
|