{"record":{"id":"412556bfd8c1e550","repo":"astrid-runtime/astrid","slug":"connection-closed-before-want-topic","errorCode":null,"errorMessage":"connection closed before {want_topic}","messagePattern":"connection closed before (.+?)","errorType":"exception","errorClass":"ReadError::ConnectionLost","httpStatus":null,"severity":"error","filePath":"crates/astrid-uplink/src/socket_client.rs","lineNumber":357,"sourceCode":"        want_topic: &str,\n        timeout: std::time::Duration,\n    ) -> std::result::Result<serde_json::Value, ReadError> {\n        let deadline = tokio::time::Instant::now()\n            .checked_add(timeout)\n            .unwrap_or_else(tokio::time::Instant::now);\n        loop {\n            let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());\n            if remaining.is_zero() {\n                return Err(ReadError::Timeout);\n            }\n            let read = tokio::time::timeout(remaining, self.read_raw_frame()).await;\n            let frame = match read {\n                Ok(Ok(Some(bytes))) => bytes,\n                // `read_raw_frame` maps a clean EOF mid-length-prefix to\n                // `Ok(None)`: the peer closed the connection (daemon restart /\n                // half-open socket), which is a connection-loss, not a timeout.\n                Ok(Ok(None)) => {\n                    return Err(ReadError::ConnectionLost(anyhow::anyhow!(\n                        \"connection closed before {want_topic}\"\n                    )));\n                },\n                // A read error (reset / broken pipe / over-large frame) is also\n                // an unusable connection.\n                Ok(Err(e)) => return Err(ReadError::ConnectionLost(e)),\n                // The outer `tokio::time::timeout` fired: the deadline elapsed\n                // with the connection still open. The broker may simply be slow.\n                Err(_) => return Err(ReadError::Timeout),\n            };\n            let raw: serde_json::Value = match serde_json::from_slice(&frame) {\n                Ok(v) => v,\n                Err(_) => continue,\n            };\n            if raw.get(\"topic\").and_then(|t| t.as_str()) == Some(want_topic) {\n                return Ok(raw);\n            }\n        }","sourceCodeStart":339,"sourceCodeEnd":375,"githubUrl":"https://github.com/astrid-runtime/astrid/blob/affd8760f44190dbdfbec23403f4c4b642c33112/crates/astrid-uplink/src/socket_client.rs#L339-L375","documentation":"astrid-uplink's `read_until_topic_typed` maps a clean EOF from `read_raw_frame` (`Ok(None)`) to `ReadError::ConnectionLost` with this message. It means the daemon (or whatever peer holds the other end of the local socket) closed the connection before sending the frame for the topic the client was waiting for — typically a daemon restart or a half-open socket. The library deliberately classifies this as connection loss rather than a timeout, so callers can distinguish 'peer went away' from 'peer is slow'.","triggerScenarios":"Calling `read_until_topic` (which delegates to `read_until_topic_typed`) when the peer sends EOF instead of the expected length-prefixed frame; also produced for hard read errors (`Ok(Err(e))`: reset, broken pipe, over-large frame) through the same `ConnectionLost` variant.","commonSituations":"The astrid daemon restarted or crashed mid-request; the socket was half-open after a suspend/network change; a proxy or socket supervisor closed the connection; the daemon rejected the client and dropped the stream before replying.","solutions":["Reconnect: drop the current client/connection and call `connect` again (the handshake will re-run against the new daemon instance).","Check whether the astrid daemon process is running and was not restarted during the call (`systemctl status` or equivalent); add reconnect-with-backoff logic around RPCs.","If it happens immediately after connect, verify the daemon version supports the topic being requested — a daemon that closes instead of replying may not implement it.","Inspect daemon logs at the time of the EOF to determine why it closed the stream."],"exampleFix":"// before: single-shot call, dies on daemon restart\nlet topic = client.read_until_topic(\"events\").await?;\n\n// after: classify and reconnect\nmatch client.read_until_topic(\"events\").await {\n    Err(ReadError::ConnectionLost(e)) => {\n        client = astrid_uplink::SocketClient::connect(&principal).await?;\n        let topic = client.read_until_topic(\"events\").await?;\n    }\n    other => other?,\n}","handlingStrategy":"retry","validationCode":"null","typeGuard":"null","tryCatchPattern":"// Rust\nmatch client.read_until_topic(topic).await {\n    Err(ReadError::ConnectionLost(e)) => {\n        // reconnect with backoff, then retry\n        client = SocketClient::connect(&principal).await?;\n        client.read_until_topic(topic).await?;\n    }\n    other => other?,\n}","preventionTips":["Wrap long-lived uplink sessions in a supervisor loop that reconnects on ConnectionLost with exponential backoff.","Monitor the daemon process (health check) so restarts are known before RPCs fail.","Keep requests short-lived; re-establish the connection after daemon upgrades instead of reusing old sockets.","Log the chained cause to distinguish clean EOF (restart) from reset/broken pipe."],"tags":["network","socket","connection-lost","rust"],"backgroundTag":"connection-closed-by-peer","analyzedSha":"affd8760f44190dbdfbec23403f4c4b642c33112","analyzedAt":"2026-09-09T21:28:12.402Z","contentChangedAt":"2026-09-09T21:28:12.402Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}