rewrite to not ack events on drop but manually ack, dont use tokio tasks and buffers

+11 -9

README.md

··· 51 51 let handle = TapHandle::spawn_default(config).await?; 52 52 53 53 // Subscribe to events 54 - let mut channel = handle.channel().await?; 54 + let (mut receiver, mut ack_sender) = handle.channel().await?; 55 55 56 - while let Ok(received) = channel.recv().await { 57 - match &received.event { 56 + while let Ok((event, ack_id)) = receiver.recv().await { 57 + match event { 58 58 Event::Record(record) => { 59 59 println!("[{:?}] {}/{}", 60 60 record.action, ··· 66 66 println!("Identity: {} -> {}", identity.did, identity.handle); 67 67 } 68 68 } 69 - // Event is auto-acknowledged when `received` is dropped 69 + // Manual acknowledgment required 70 + ack_sender.ack(ack_id).await?; 70 71 } 71 72 72 73 Ok(()) ··· 164 165 165 166 ### Working with Events 166 167 167 - Events are automatically acknowledged when dropped: 168 + Events must be manually acknowledged: 168 169 169 170 ```rust 170 171 use tapped::{Event, RecordAction}; 171 172 172 - let mut channel = client.channel().await?; 173 + let (mut receiver, mut ack_sender) = client.channel().await?; 173 174 174 - while let Ok(received) = channel.recv().await { 175 - match &received.event { 175 + while let Ok((event, ack_id)) = receiver.recv().await { 176 + match event { 176 177 Event::Record(record) => { 177 178 match record.action { 178 179 RecordAction::Create => { ··· 193 194 println!("{} is now @{}", identity.did, identity.handle); 194 195 } 195 196 } 196 - // Ack sent automatically here when `received` goes out of scope 197 + // Ack must be sent manually 198 + ack_sender.ack(ack_id).await?; 197 199 } 198 200 ``` 199 201

+10 -6

standard-site-sync/src/main.rs

··· 67 67 client.health().await?; 68 68 info!("Tap is healthy!"); 69 69 70 - let mut receiver = client.channel().await?; 70 + let (mut receiver, mut ack_sender) = client.channel().await?; 71 71 info!("Connected! Waiting for events..."); 72 72 73 73 // In-memory cache - load from disk if available ··· 83 83 84 84 loop { 85 85 match receiver.recv().await { 86 - Ok(received) => { 87 - if let Event::Record(ref record_event) = *received { 86 + Ok((event, ack_id)) => { 87 + if let Event::Record(ref record_event) = event { 88 88 // Track live vs backfill 89 89 if record_event.live { 90 90 live_count += 1; ··· 96 96 write_output_files(&cache)?; 97 97 98 98 // Periodically show event source breakdown 99 - if (live_count + backfill_count).is_multiple_of(10) { 99 + if (live_count + backfill_count) % 10 == 0 { 100 100 info!( 101 101 "[Stats] Live events: {}, Backfill events: {}", 102 102 live_count, backfill_count 103 103 ); 104 104 } 105 - } else if let Event::Identity(ref identity_event) = *received { 105 + } else if let Event::Identity(ref identity_event) = event { 106 106 info!( 107 107 "[IDENTITY] {} -> {} (active: {})", 108 108 identity_event.did, identity_event.handle, identity_event.is_active 109 109 ); 110 110 } 111 - // Event is automatically acked when `received` is dropped here 111 + 112 + if let Err(e) = ack_sender.ack(ack_id).await { 113 + error!("Failed to ack event: {}", e); 114 + break; 115 + } 112 116 } 113 117 Err(e) => { 114 118 eprintln!("Error receiving event: {}", e);

+55 -133

tapped/src/channel.rs

··· 1 1 //! WebSocket event channel and receiver. 2 2 3 3 use serde::Serialize; 4 - use tokio::sync::mpsc; 5 4 use sockudo_ws::{Message, Http1, Config, Stream as WsTransportStream, SplitWriter, SplitReader}; 6 5 use sockudo_ws::client::WebSocketClient; 7 - use bytes::Bytes; 8 6 use url::Url; 9 7 10 8 use crate::types::RawEvent; ··· 13 11 type WsSink = SplitWriter<WsTransportStream<Http1>>; 14 12 type WsSource = SplitReader<WsTransportStream<Http1>>; 15 13 14 + /// Opaque identifier for an event to be acknowledged. 15 + #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] 16 + pub struct AckId(u64); 17 + 18 + /// Sender for event acknowledgments. 19 + pub struct AckSender { 20 + write: WsSink, 21 + } 22 + 23 + impl AckSender { 24 + /// Send an acknowledgment for an event. 25 + pub async fn ack(&mut self, id: AckId) -> Result<()> { 26 + #[derive(Serialize)] 27 + struct AckMessage { 28 + #[serde(rename = "type")] 29 + type_: &'static str, 30 + id: u64, 31 + } 32 + 33 + let msg = AckMessage { type_: "ack", id: id.0 }; 34 + let json = serde_json::to_string(&msg)?; 35 + 36 + self.write.send(Message::text(json)).await.map_err(|e| Error::WebSocket(Box::new(e))) 37 + } 38 + } 39 + 16 40 /// Receiver for events from a tap WebSocket channel. 17 41 /// 18 42 /// Events are received via the [`recv`](EventReceiver::recv) method. 19 - /// Acknowledgments are sent automatically when events are dropped. 43 + /// Acknowledgments must be sent manually using the [`AckSender`] returned by [`TapClient::channel()`](crate::TapClient::channel). 20 44 /// 21 45 /// This type does not implement auto-reconnection. If the connection 22 46 /// closes, `recv()` will return an error and you must create a new 23 47 /// `EventReceiver` via [`TapClient::channel()`](crate::TapClient::channel). 24 48 pub struct EventReceiver { 25 - event_rx: mpsc::Receiver<EventWithAck>, 26 - _ack_tx: mpsc::Sender<u64>, 27 - } 28 - 29 - struct EventWithAck { 30 - event: Bytes, 31 - ack_tx: mpsc::Sender<u64>, 32 - } 33 - 34 - struct AckGuard { 35 - id: u64, 36 - ack_tx: Option<mpsc::Sender<u64>>, 37 - } 38 - 39 - impl Drop for AckGuard { 40 - fn drop(&mut self) { 41 - if let Some(tx) = self.ack_tx.take() { 42 - // Fire and forget - if the channel is closed, we can't ack anyway 43 - let id = self.id; 44 - tokio::spawn(async move { 45 - let _ = tx.send(id).await; 46 - }); 47 - } 48 - } 49 - } 50 - 51 - /// Wrapper around Event that includes the ack trigger. 52 - pub struct ReceivedEvent { 53 - pub event: Event, 54 - _ack_guard: AckGuard, 55 - } 56 - 57 - impl std::ops::Deref for ReceivedEvent { 58 - type Target = Event; 59 - 60 - fn deref(&self) -> &Self::Target { 61 - &self.event 62 - } 49 + read: WsSource, 63 50 } 64 51 65 52 impl EventReceiver { 66 53 /// Connect to a tap WebSocket channel. 67 - pub(crate) async fn connect(base_url: &Url, admin_password: Option<&str>) -> Result<Self> { 54 + pub(crate) async fn connect(base_url: &Url, admin_password: Option<&str>) -> Result<(Self, AckSender)> { 68 55 let mut ws_url = base_url.clone(); 69 56 match ws_url.scheme() { 70 57 "http" => ws_url.set_scheme("ws").unwrap(), ··· 122 109 123 110 let (read, write) = ws_stream.split(); 124 111 125 - // Buffer size increased to 2048 to prevent TCP window clamping during brief processing spikes 126 - let (event_tx, event_rx) = mpsc::channel(2048); 127 - let (ack_tx, ack_rx) = mpsc::channel(1000); 128 - 129 - let ack_tx_clone = ack_tx.clone(); 130 - tokio::spawn(async move { 131 - Self::writer_task(write, ack_rx).await; 132 - }); 133 - 134 - tokio::spawn(async move { 135 - Self::reader_task(read, event_tx, ack_tx_clone).await; 136 - }); 137 - 138 - Ok(Self { 139 - event_rx, 140 - _ack_tx: ack_tx, 141 - }) 112 + Ok(( 113 + Self { read }, 114 + AckSender { write }, 115 + )) 142 116 } 143 117 144 118 /// Receive the next event. 145 119 /// 146 - /// Returns the event wrapped in a [`ReceivedEvent`] that automatically 147 - /// sends an acknowledgment when dropped. 120 + /// Returns the event and an opaque ID that must be passed to [`AckSender::ack`] 121 + /// to acknowledge the event. 148 122 /// 149 123 /// # Errors 150 124 /// 151 125 /// Returns [`Error::ChannelClosed`] if the WebSocket connection closes. 152 - pub async fn recv(&mut self) -> Result<ReceivedEvent> { 126 + pub async fn recv(&mut self) -> Result<(Event, AckId)> { 153 127 loop { 154 - match self.event_rx.recv().await { 155 - Some(event_with_ack) => { 156 - let json = event_with_ack.event; 157 - let json_str = std::str::from_utf8(&json).expect("must be utf8"); 128 + match self.read.next().await { 129 + Some(Ok(Message::Text(event_bytes))) => { 130 + let event_bytes = bytes::Bytes::from(event_bytes); 131 + let json_str = match std::str::from_utf8(&event_bytes) { 132 + Ok(s) => s, 133 + Err(e) => { 134 + tracing::warn!("Failed to parse event as utf8: {}", e); 135 + continue; 136 + } 137 + }; 138 + 158 139 let raw = match serde_json::from_str::<RawEvent>(json_str) { 159 140 Ok(raw) => raw, 160 141 Err(e) => { 161 - tracing::warn!("Failed to parse event: {}", e); 142 + tracing::warn!("Failed to parse event json: {}", e); 162 143 continue; 163 144 } 164 145 }; 165 146 166 - if let Some(event) = raw.into_event(json.clone()) { 147 + if let Some(event) = raw.into_event(event_bytes.clone()) { 167 148 let id = event.id(); 168 - break Ok(ReceivedEvent { 169 - event, 170 - _ack_guard: AckGuard { 171 - id, 172 - ack_tx: Some(event_with_ack.ack_tx), 173 - }, 174 - }); 175 - } 176 - } 177 - None => break Err(Error::ChannelClosed), 178 - } 179 - } 180 - } 181 - 182 - /// Writer task: sends ack messages to the WebSocket. 183 - async fn writer_task(mut write: WsSink, mut ack_rx: mpsc::Receiver<u64>) { 184 - #[derive(Serialize)] 185 - struct AckMessage { 186 - #[serde(rename = "type")] 187 - type_: &'static str, 188 - id: u64, 189 - } 190 - 191 - while let Some(id) = ack_rx.recv().await { 192 - let msg = AckMessage { type_: "ack", id }; 193 - let json = match serde_json::to_string(&msg) { 194 - Ok(j) => j, 195 - Err(e) => { 196 - tracing::warn!("Failed to serialize ack: {}", e); 197 - continue; 198 - } 199 - }; 200 - 201 - if let Err(e) = write.send(Message::text(json)).await { 202 - tracing::warn!("Failed to send ack: {}", e); 203 - break; 204 - } 205 - } 206 - } 207 - 208 - /// Reader task: reads events from WebSocket and sends to channel. 209 - async fn reader_task( 210 - mut read: WsSource, 211 - event_tx: mpsc::Sender<EventWithAck>, 212 - ack_tx: mpsc::Sender<u64>, 213 - ) { 214 - while let Some(msg_result) = read.next().await { 215 - match msg_result { 216 - Ok(Message::Text(event)) => { 217 - let event_with_ack = EventWithAck { 218 - event, 219 - ack_tx: ack_tx.clone(), 220 - }; 221 - if event_tx.send(event_with_ack).await.is_err() { 222 - break; 149 + return Ok((event, AckId(id))); 223 150 } 224 151 } 225 - Ok(Message::Close(_)) => { 226 - break; 227 - } 228 - Ok(_) => { 229 - // Ignore ping/pong/binary 230 - } 231 - Err(_) => { 232 - break; 233 - } 152 + Some(Ok(Message::Close(_))) => return Err(Error::ChannelClosed), 153 + Some(Ok(_)) => continue, // Ping/Pong/Binary 154 + Some(Err(_)) => return Err(Error::ChannelClosed), 155 + None => return Err(Error::ChannelClosed), 234 156 } 235 157 } 236 158 } 237 - } 159 + }

+7 -6

tapped/src/client.rs

··· 337 337 338 338 /// Connect to the WebSocket event channel. 339 339 /// 340 - /// Returns an [`EventReceiver`] for receiving events. Events are 341 - /// automatically acknowledged when dropped. 340 + /// Returns an [`EventReceiver`] for receiving events and an [`AckSender`] for 341 + /// acknowledging them. Events must be manually acknowledged using [`AckSender::ack`]. 342 342 /// 343 343 /// # Example 344 344 /// ··· 347 347 /// use tapped::TapClient; 348 348 /// 349 349 /// let client = TapClient::new("http://localhost:2480")?; 350 - /// let mut receiver = client.channel().await?; 350 + /// let (mut receiver, mut ack_sender) = client.channel().await?; 351 351 /// 352 - /// while let Ok(event) = receiver.recv().await { 353 - /// // Event is automatically acknowledged when dropped 352 + /// while let Ok((event, ack_id)) = receiver.recv().await { 353 + /// // Process event... 354 + /// ack_sender.ack(ack_id).await?; 354 355 /// } 355 356 /// # Ok(()) 356 357 /// # } 357 358 /// ``` 358 - pub async fn channel(&self) -> Result<EventReceiver> { 359 + pub async fn channel(&self) -> Result<(EventReceiver, crate::channel::AckSender)> { 359 360 EventReceiver::connect(&self.base_url, self.admin_password.as_deref()).await 360 361 } 361 362 }

+3 -2

tapped/src/handle.rs

··· 32 32 /// // Use the client methods directly on the handle 33 33 /// handle.health().await?; 34 34 /// 35 - /// let mut channel = handle.channel().await?; 36 - /// while let Ok(event) = channel.recv().await { 35 + /// let (mut receiver, mut ack_sender) = handle.channel().await?; 36 + /// while let Ok((event, ack_id)) = receiver.recv().await { 37 37 /// // Handle event 38 + /// ack_sender.ack(ack_id).await?; 38 39 /// } 39 40 /// 40 41 /// Ok(())

+7 -6

tapped/src/lib.rs

··· 10 10 //! 11 11 //! - Connect to an existing tap instance or spawn one as a subprocess 12 12 //! - Strongly-typed configuration with builder pattern 13 - //! - Async event streaming with automatic acknowledgment 13 + //! - Async event streaming with manual acknowledgment 14 14 //! - Full HTTP API coverage for repo management and statistics 15 15 //! 16 16 //! ## Example ··· 30 30 //! client.add_repos(&["did:plc:example1234567890abc"]).await?; 31 31 //! 32 32 //! // Stream events 33 - //! let mut receiver = client.channel().await?; 34 - //! while let Ok(event) = receiver.recv().await { 35 - //! // Event is automatically acknowledged when dropped 33 + //! let (mut receiver, mut ack_sender) = client.channel().await?; 34 + //! while let Ok((event, ack_id)) = receiver.recv().await { 35 + //! // Process event... 36 + //! ack_sender.ack(ack_id).await?; 36 37 //! } 37 38 //! 38 39 //! Ok(()) ··· 47 48 mod process; 48 49 mod types; 49 50 50 - pub use channel::{EventReceiver, ReceivedEvent}; 51 + pub use channel::{EventReceiver, AckSender, AckId}; 51 52 pub use client::TapClient; 52 53 pub use config::{LogLevel, TapConfig, TapConfigBuilder}; 53 54 pub use error::Error; ··· 59 60 }; 60 61 61 62 /// A specialised Result type for tapped operations. 62 - pub type Result<T> = std::result::Result<T, Error>; 63 + pub type Result<T> = std::result::Result<T, Error>;

+1 -1

tapped/tests/integration.rs

··· 149 149 .await 150 150 .expect("Failed to spawn tap"); 151 151 152 - let _channel = handle.channel().await.expect("channel connection failed"); 152 + let (_receiver, _ack_sender) = handle.channel().await.expect("channel connection failed"); 153 153 } 154 154 155 155 #[tokio::test]

Configure Feed

Configure Feed