{"record":{"id":"0aa832018d3597ef","repo":"nautechsystems/nautilus_trader","slug":"failed-to-write-pool-fee-collect-data-e","errorCode":null,"errorMessage":"Failed to write pool fee collect data: {e}","messagePattern":"Failed to write pool fee collect data: (.+?)","errorType":"exception","errorClass":"anyhow::Error","httpStatus":null,"severity":"error","filePath":"crates/adapters/blockchain/src/cache/copy.rs","lineNumber":762,"sourceCode":"        let owner_bytes = collect.owner.to_string().as_bytes().to_vec();\n        row_data.write_all(&(owner_bytes.len() as i32).to_be_bytes())?;\n        row_data.write_all(&owner_bytes)?;\n\n        write_copy_numeric(&mut row_data, collect.amount0);\n        write_copy_numeric(&mut row_data, collect.amount1);\n\n        let tick_lower_bytes = collect.tick_lower.to_be_bytes();\n        row_data.write_all(&(tick_lower_bytes.len() as i32).to_be_bytes())?;\n        row_data.write_all(&tick_lower_bytes)?;\n\n        let tick_upper_bytes = collect.tick_upper.to_be_bytes();\n        row_data.write_all(&(tick_upper_bytes.len() as i32).to_be_bytes())?;\n        row_data.write_all(&tick_upper_bytes)?;\n\n        copy_in\n            .send(row_data)\n            .await\n            .map_err(|e| anyhow::anyhow!(\"Failed to write pool fee collect data: {e}\"))?;\n        Ok(())\n    }\n\n    /// Writes a single token in PostgreSQL binary format.\n    ///\n    /// Each row in binary format consists of:\n    /// - 2-byte field count\n    /// - For each field: 4-byte length followed by data (or -1 for NULL)\n    async fn write_token_binary(\n        &self,\n        copy_in: &mut sqlx::postgres::PgCopyIn<sqlx::pool::PoolConnection<sqlx::Postgres>>,\n        chain_id: u32,\n        token: &Token,\n    ) -> anyhow::Result<()> {\n        use std::io::Write;\n        let mut row_data = Vec::new();\n\n        row_data.write_all(&5u16.to_be_bytes())?;","sourceCodeStart":744,"sourceCodeEnd":780,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/adapters/blockchain/src/cache/copy.rs#L744-L780","documentation":"write_pool_fee_collect_binary serializes one pool fee collect row in PostgreSQL COPY binary format and sends it through the copy_in sink. This error wraps the sink's send error when the row bytes cannot be delivered to the server. It is per-row; the underlying cause is usually a connection problem or a server-side rejection of the COPY stream.","triggerScenarios":"copy_in.send(row_data) fails inside write_pool_fee_collect_binary while copy_pool_collects streams rows — connection closed, server aborted the COPY, or a prior row caused the channel to error.","commonSituations":"Database restart or failover during backfill; statement/network timeout on a long COPY; schema drift making the binary encoding of tick_upper or amount fields invalid for the target column type.","solutions":["Inspect the chained {e} source for the concrete sqlx/postgres error (connection vs encoding)","Confirm the table schema still matches the binary field order and types used by write_pool_fee_collect_binary","Check network stability and increase TCP keepalive / remove short timeouts for bulk loads","Retry the whole copy_pool_collects operation after the connection recovers"],"exampleFix":"// before: unbounded loop without backpressure handling can stall/fail on dropped connection\nfor c in collects { write_pool_fee_collect_binary(&mut copy_in, c).await?; }\n// after: chunk and retry on transient errors\nfor chunk in collects.chunks(10_000) {\n    for c in chunk {\n        write_pool_fee_collect_binary(&mut copy_in, c).await\n            .map_err(|e| e.context(\"pool fee collect row\"))?;\n    }\n}","handlingStrategy":"retry","validationCode":"// pre-validate row against expected schema before streaming\nfn validate_collect(c: &PoolFeeCollect) -> Result<(), String> {\n    if c.amount.is_negative() { return Err(\"negative collect amount\".into()); }\n    Ok(())\n}","typeGuard":null,"tryCatchPattern":"for chunk in collects.chunks(5_000) {\n    if let Err(e) = copy_collect_chunk(db, chain, chunk).await {\n        if is_connection_error(&e) {\n            tokio::time::sleep(Duration::from_secs(2)).await;\n            copy_collect_chunk(db, chain, chunk).await?;\n        } else { return Err(e); }\n    }\n}","preventionTips":["Validate collect values (ticks, amounts) before serializing to binary format","Keep schema migrations in lockstep with the adapter's binary row encoding","Use TCP keepalive on the Postgres connection for long-running syncs","Chunk large row sets so failures are easier to isolate and retry"],"tags":["database","postgres","copy","binary-format"],"backgroundTag":"database-write-failed","analyzedSha":"18893faf8b356be3320add8de2f861b0b647cf06","analyzedAt":"2026-09-08T20:49:34.690Z","contentChangedAt":"2026-09-08T20:49:34.690Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}