Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 27 additions & 11 deletions src/engine/postgres_psql.rs
Original file line number Diff line number Diff line change
Expand Up @@ -530,11 +530,23 @@ impl PSQL {

let status = child.wait().await?;

let writer_err = match writer_result {
Err(join_err) => Some(std::io::Error::other(join_err)), // the task panicked
Ok(Err(e)) => Some(e),
Ok(Ok(())) => None,
};

// Takes precedence over psql's own status: an aborted render can leave
// psql exiting 0 (e.g. a quiet rollback on early EOF).
writer_result
.map_err(|e| EngineError::Io(std::io::Error::new(std::io::ErrorKind::Other, e)))
.and_then(|r| r.map_err(EngineError::Io))?;
// psql exiting 0 (e.g. a quiet rollback on early EOF). Unless the pipe
// broke, which means psql had already exited: then its error is the
// cause and ours only the symptom.
if let Some(e) = writer_err {
let psql_died_first = e.kind() == std::io::ErrorKind::BrokenPipe && !status.success();
if !psql_died_first {
return Err(EngineError::Io(e));
}
// Dropped: the status check below reports psql's error instead.
}
results_result?;
diagnostics_result?;

Expand Down Expand Up @@ -984,13 +996,17 @@ impl PSQL {
}
Err(EngineError::Io(e)) => {
// Writer failed (e.g. an unresolved secret), not psql itself.
// Still record a Failure row — SQL may already have run.
(
MigrationStatus::Failure,
Some(
anyhow::Error::from(e).context("failed while streaming migration SQL"),
),
)
// Still record a Failure row — SQL may already have run, and
// psql may have reported on what it did receive.
let base = anyhow::Error::from(e);
let err = match tail_summary(&diagnostics.tail, diagnostics.dropped) {
Some(stderr) => base.context(format!(
"failed while streaming migration SQL; psql reported: {}",
stderr
)),
None => base.context("failed while streaming migration SQL"),
};
(MigrationStatus::Failure, Some(err))
}
// Only `run_script` produces this; `execute_with_writer` never does.
Err(EngineError::Unavailable { message }) => {
Expand Down
12 changes: 10 additions & 2 deletions src/template.rs
Original file line number Diff line number Diff line change
Expand Up @@ -425,8 +425,16 @@ impl StreamingGeneration {
pub fn into_script_source(self) -> (crate::engine::ScriptSource, Arc<SecretsRepository>) {
let secrets = Arc::clone(&self.secrets);
let script = Box::new(move |writer: &mut dyn std::io::Write| {
self.render_to_writer(writer)
.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))
self.render_to_writer(writer).map_err(|e| {
// minijinja buries the writer's io::Error under WriteFailure;
// carry its kind back out so callers can spot a broken pipe.
let kind = e
.chain()
.find_map(|c| c.downcast_ref::<std::io::Error>())
.map(|io| io.kind())
.unwrap_or(std::io::ErrorKind::Other);
std::io::Error::new(kind, e)
})
});
(script, secrets)
}
Expand Down
59 changes: 59 additions & 0 deletions tests/integration_postgres.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2492,3 +2492,62 @@ async fn test_run_script_does_not_hang_when_a_sink_fails() -> Result<()> {

Ok(())
}

/// psql exits on a failed statement under ON_ERROR_STOP. If the script is still
/// being written, the write fails with `BrokenPipe` — a symptom that must not
/// displace psql's own report.
///
/// Writing until the pipe breaks is deliberate: a fixed payload is a race, and
/// under `docker exec -i` even ~760KB won 2 runs in 3.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore]
async fn test_broken_pipe_does_not_mask_the_psql_error() -> Result<()> {
use spawn_db::engine::ScriptSource;

require_postgres()?;

let helper =
IntegrationTestHelper::new("test_broken_pipe_does_not_mask_the_psql_error", None).await?;
let config = helper.migration_helper.load_config().await?;
let engine = config.new_engine().await?;

let script: ScriptSource = Box::new(|writer| {
writer.write_all(b"SELECT this_does_not_exist();\n")?;
// A full line of dashes, so anything psql does read stays a comment.
let padding = [b'-'; 4096];
loop {
writer.write_all(&padding)?;
writer.write_all(b"\n")?;
}
});

let outcome = tokio::time::timeout(
std::time::Duration::from_secs(30),
engine.migration_apply(
"20260101000000-broken-pipe",
script,
"deadbeef".to_string(),
None,
"default",
false,
),
)
.await
.map_err(|_| anyhow!("migration_apply hung: the writer was never unblocked"))?;

let err = outcome.expect_err("a failing statement should surface as an error");
let chain = format!("{:?}", anyhow::Error::new(err));

assert!(
chain.contains("this_does_not_exist") && chain.contains("psql exited with code"),
"apply must report psql's error, not just that the write failed, got: {}",
chain
);
assert!(
!chain.contains("failed while streaming migration SQL"),
"psql died first, so this is not a writer failure, got: {}",
chain
);

Ok(())
}
Loading