rfc5321 client: fixup pipelining error handling

98a5479ce9 changed the logic around
pipelining error handling in a subtle way: it switched from sequentially
considering both the transport and protocol errors for each of the
pipelined commands to eager consideration of transport errors, before
considering any transport errors.

In a situation where the remote host issues a 421 response to the MAIL
FROM that is part of our pipeline of (RSET, MAIL FROM, RCPT TO, DATA),
rather than reporting a 421 response to MAIl FROM, we'd end up bubbling
up the NotConnected error from RCPT TO, leading to the miscategorization
and so on that was improved in caadc0e7e8

This commit fixes up this situation by:

* Ensuring that we pad out the number of results from pipeline_commands
  to match the number of commands, synthesizing NotConnected errors
  for the remaining slots if we experience a transport error.

* Check for any transport error and adjust the logic to essentially
  prefer handling a protocol error from an earlier stage, rather
  than taking the transport error.  This allows the transport-success
  (eg: peer didn't disconnect) case to continue to defer acting upon
  any protocol errors until we have the full set of protocol responses
  back, so that we can handle the obscure but important DATA case that
  was the original intent behind the changes in
  98a5479ce9

crates/integration-tests/src/test/disconnect_in_mail_from.rs

@@ -28,6 +28,23 @@ async fn disconnect_in_mail_from() -> anyhow::Result<()> {
         .await;
 
     daemon.stop_both().await?;
+
+    let logs = daemon.source.collect_logs()?;
+    k9::snapshot!(
+        logs.last().unwrap().response.clone(),
+        r#"
+Response {
+    code: 421,
+    enhanced_code: None,
+    content: "disconnecting 421-disconnect-me",
+    command: Some(
+        "MAIL FROM:<421-disconnect-me@example.com>\r
+",
+    ),
+}
+"#
+    );
+
     let delivery_summary = daemon.dump_logs()?;
     k9::snapshot!(
         delivery_summary,

crates/rfc5321/src/client.rs

@@ -407,6 +407,10 @@ impl SmtpClient {
         for cmd in &commands {
             if let Err(err) = self.write_command_request(cmd, pipeline).await {
                 results.push(Err(err.into()));
+                while results.len() < commands.len() {
+                    // Synthesize failures for the remaining commands
+                    results.push(Err(ClientError::NotConnected));
+                }
                 return results;
             }
             if !pipeline {
@@ -692,16 +696,34 @@ impl SmtpClient {
             ])
             .await;
 
-        if responses.is_empty() {
-            // Should be impossible to get here really, but if we do,
-            // assume that we aren't connected
-            return Err(ClientError::NotConnected);
-        }
+        // This is a little awkward. We want to handle the RFC 2090 3.1 case
+        // below, which requires deferring checking the actual response codes
+        // until later, but we must also handle the case where we had a hard
+        // transport error partway through pipelining.
+        // So we set a flag for that case and will then "eagerly", wrt. the
+        // RFC 2090 3.1 logic, evaluate the SMTP response codes, so that we
+        // can propagate the correct error disposition up to the caller.
+        let is_err = responses.iter().any(|r| r.is_err());
 
         let rset_resp = responses.remove(0)?;
+        if is_err && rset_resp.code != 250 {
+            return Err(ClientError::Rejected(rset_resp));
+        }
+
         let mail_resp = responses.remove(0)?;
+        if is_err && mail_resp.code != 250 {
+            return Err(ClientError::Rejected(mail_resp));
+        }
+
         let rcpt_resp = responses.remove(0)?;
+        if is_err && rcpt_resp.code != 250 {
+            return Err(ClientError::Rejected(rcpt_resp));
+        }
+
         let data_resp = responses.remove(0)?;
+        if is_err && data_resp.code != 354 {
+            return Err(ClientError::Rejected(data_resp));
+        }
 
         if data_resp.code == 354
             && (rset_resp.code != 250 || mail_resp.code != 250 || rcpt_resp.code != 250)