diff --git a/src/test/regress/expected/failure_multi_dml.out b/src/test/regress/expected/failure_multi_dml.out
new file mode 100644
index 000000000..abee6d84e
--- /dev/null
+++ b/src/test/regress/expected/failure_multi_dml.out
@@ -0,0 +1,304 @@
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy 
+-----------
+ 
+(1 row)
+
+SET citus.shard_count = 2;
+SET citus.shard_replication_factor = 1; -- one shard per worker
+SET citus.next_shard_id TO 103400;
+ALTER SEQUENCE pg_catalog.pg_dist_placement_placementid_seq RESTART 100;
+CREATE TABLE dml_test (id integer, name text);
+SELECT create_distributed_table('dml_test', 'id');
+ create_distributed_table 
+--------------------------
+ 
+(1 row)
+
+COPY dml_test FROM STDIN WITH CSV;
+SELECT citus.clear_network_traffic();
+ clear_network_traffic 
+-----------------------
+ 
+(1 row)
+
+---- test multiple statements spanning multiple shards,
+---- at each significant point. These transactions are 2pc
+-- fail at DELETE
+SELECT citus.mitmproxy('conn.onQuery(query="^DELETE").kill()');
+ mitmproxy 
+-----------
+ 
+(1 row)
+
+BEGIN;
+DELETE FROM dml_test WHERE id = 1;
+ERROR:  server closed the connection unexpectedly
+	This probably means the server terminated abnormally
+	before or while processing the request.
+CONTEXT:  while executing command on localhost:9060
+DELETE FROM dml_test WHERE id = 2;
+ERROR:  current transaction is aborted, commands ignored until end of transaction block
+INSERT INTO dml_test VALUES (5, 'Epsilon');
+ERROR:  current transaction is aborted, commands ignored until end of transaction block
+UPDATE dml_test SET name = 'alpha' WHERE id = 1;
+ERROR:  current transaction is aborted, commands ignored until end of transaction block
+UPDATE dml_test SET name = 'gamma' WHERE id = 3;
+ERROR:  current transaction is aborted, commands ignored until end of transaction block
+COMMIT;
+--- shouldn't see any changes performed in failed transaction
+SELECT * FROM dml_test ORDER BY id ASC;
+ id | name  
+----+-------
+  1 | Alpha
+  2 | Beta
+  3 | Gamma
+  4 | Delta
+(4 rows)
+
+-- fail at INSERT
+SELECT citus.mitmproxy('conn.onQuery(query="^INSERT").kill()');
+ mitmproxy 
+-----------
+ 
+(1 row)
+
+BEGIN;
+DELETE FROM dml_test WHERE id = 1;
+DELETE FROM dml_test WHERE id = 2;
+INSERT INTO dml_test VALUES (5, 'Epsilon');
+ERROR:  server closed the connection unexpectedly
+	This probably means the server terminated abnormally
+	before or while processing the request.
+CONTEXT:  while executing command on localhost:9060
+UPDATE dml_test SET name = 'alpha' WHERE id = 1;
+ERROR:  current transaction is aborted, commands ignored until end of transaction block
+UPDATE dml_test SET name = 'gamma' WHERE id = 3;
+ERROR:  current transaction is aborted, commands ignored until end of transaction block
+COMMIT;
+--- shouldn't see any changes before failed INSERT
+SELECT * FROM dml_test ORDER BY id ASC;
+ id | name  
+----+-------
+  1 | Alpha
+  2 | Beta
+  3 | Gamma
+  4 | Delta
+(4 rows)
+
+-- fail at UPDATE
+SELECT citus.mitmproxy('conn.onQuery(query="^UPDATE").kill()');
+ mitmproxy 
+-----------
+ 
+(1 row)
+
+BEGIN;
+DELETE FROM dml_test WHERE id = 1;
+DELETE FROM dml_test WHERE id = 2;
+INSERT INTO dml_test VALUES (5, 'Epsilon');
+UPDATE dml_test SET name = 'alpha' WHERE id = 1;
+ERROR:  server closed the connection unexpectedly
+	This probably means the server terminated abnormally
+	before or while processing the request.
+CONTEXT:  while executing command on localhost:9060
+UPDATE dml_test SET name = 'gamma' WHERE id = 3;
+ERROR:  current transaction is aborted, commands ignored until end of transaction block
+COMMIT;
+--- shouldn't see any changes after failed UPDATE
+SELECT * FROM dml_test ORDER BY id ASC;
+ id | name  
+----+-------
+  1 | Alpha
+  2 | Beta
+  3 | Gamma
+  4 | Delta
+(4 rows)
+
+-- fail at PREPARE TRANSACTION
+SELECT citus.mitmproxy('conn.onQuery(query="^PREPARE TRANSACTION").kill()');
+ mitmproxy 
+-----------
+ 
+(1 row)
+
+-- hide the error message (it has the PID)...
+-- we'll test for the txn side-effects to ensure it didn't run
+SET client_min_messages TO FATAL;
+BEGIN;
+DELETE FROM dml_test WHERE id = 1;
+DELETE FROM dml_test WHERE id = 2;
+INSERT INTO dml_test VALUES (5, 'Epsilon');
+UPDATE dml_test SET name = 'alpha' WHERE id = 1;
+UPDATE dml_test SET name = 'gamma' WHERE id = 3;
+COMMIT;
+SET client_min_messages TO DEFAULT;
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy 
+-----------
+ 
+(1 row)
+
+SELECT shardid FROM pg_dist_shard_placement WHERE shardstate = 3;
+ shardid 
+---------
+(0 rows)
+
+SELECT recover_prepared_transactions();
+ recover_prepared_transactions 
+-------------------------------
+                             0
+(1 row)
+
+-- shouldn't see any changes after failed PREPARE
+SELECT * FROM dml_test ORDER BY id ASC;
+ id | name  
+----+-------
+  1 | Alpha
+  2 | Beta
+  3 | Gamma
+  4 | Delta
+(4 rows)
+
+-- fail at COMMIT
+SELECT citus.mitmproxy('conn.onQuery(query="^COMMIT").kill()');
+ mitmproxy 
+-----------
+ 
+(1 row)
+
+-- hide the error message (it has the PID)...
+-- we'll test for the txn side-effects to ensure it didn't run
+SET client_min_messages TO FATAL;
+BEGIN;
+DELETE FROM dml_test WHERE id = 1;
+DELETE FROM dml_test WHERE id = 2;
+INSERT INTO dml_test VALUES (5, 'Epsilon');
+UPDATE dml_test SET name = 'alpha' WHERE id = 1;
+UPDATE dml_test SET name = 'gamma' WHERE id = 3;
+COMMIT;
+SET client_min_messages TO DEFAULT;
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy 
+-----------
+ 
+(1 row)
+
+SELECT shardid FROM pg_dist_shard_placement WHERE shardstate = 3;
+ shardid 
+---------
+(0 rows)
+
+SELECT recover_prepared_transactions();
+ recover_prepared_transactions 
+-------------------------------
+                             1
+(1 row)
+
+-- should see changes, because of txn recovery
+SELECT * FROM dml_test ORDER BY id ASC;
+ id |  name   
+----+---------
+  3 | gamma
+  4 | Delta
+  5 | Epsilon
+(3 rows)
+
+-- drop table and recreate with different replication/sharding
+DROP TABLE dml_test;
+SET citus.shard_count = 1;
+SET citus.shard_replication_factor = 2; -- two placements
+CREATE TABLE dml_test (id integer, name text);
+SELECT create_distributed_table('dml_test', 'id');
+ create_distributed_table 
+--------------------------
+ 
+(1 row)
+
+COPY dml_test FROM STDIN WITH CSV;
+---- test multiple statements against a single shard, but with two placements
+-- fail at COMMIT (actually COMMIT this time, as no 2pc in use)
+SELECT citus.mitmproxy('conn.onQuery(query="^COMMIT").kill()');
+ mitmproxy 
+-----------
+ 
+(1 row)
+
+BEGIN;
+DELETE FROM dml_test WHERE id = 1;
+DELETE FROM dml_test WHERE id = 2;
+INSERT INTO dml_test VALUES (5, 'Epsilon');
+UPDATE dml_test SET name = 'alpha' WHERE id = 1;
+UPDATE dml_test SET name = 'gamma' WHERE id = 3;
+COMMIT;
+WARNING:  connection not open
+CONTEXT:  while executing command on localhost:9060
+WARNING:  failed to commit transaction on localhost:9060
+WARNING:  connection not open
+CONTEXT:  while executing command on localhost:9060
+--- should see all changes, but they only went to one placement (other is unhealthy)
+SELECT * FROM dml_test ORDER BY id ASC;
+ id |  name   
+----+---------
+  3 | gamma
+  4 | Delta
+  5 | Epsilon
+(3 rows)
+
+SELECT shardid FROM pg_dist_shard_placement WHERE shardstate = 3;
+ shardid 
+---------
+  103402
+(1 row)
+
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy 
+-----------
+ 
+(1 row)
+
+-- drop table and recreate as reference table
+DROP TABLE dml_test;
+SET citus.shard_count = 2;
+SET citus.shard_replication_factor = 1;
+CREATE TABLE dml_test (id integer, name text);
+SELECT create_reference_table('dml_test');
+ create_reference_table 
+------------------------
+ 
+(1 row)
+
+COPY dml_test FROM STDIN WITH CSV;
+-- fail at COMMIT (by failing to PREPARE)
+SELECT citus.mitmproxy('conn.onQuery(query="^PREPARE").kill()');
+ mitmproxy 
+-----------
+ 
+(1 row)
+
+BEGIN;
+DELETE FROM dml_test WHERE id = 1;
+DELETE FROM dml_test WHERE id = 2;
+INSERT INTO dml_test VALUES (5, 'Epsilon');
+UPDATE dml_test SET name = 'alpha' WHERE id = 1;
+UPDATE dml_test SET name = 'gamma' WHERE id = 3;
+COMMIT;
+ERROR:  connection not open
+CONTEXT:  while executing command on localhost:9060
+--- shouldn't see any changes after failed COMMIT
+SELECT * FROM dml_test ORDER BY id ASC;
+ id | name  
+----+-------
+  1 | Alpha
+  2 | Beta
+  3 | Gamma
+  4 | Delta
+(4 rows)
+
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy 
+-----------
+ 
+(1 row)
+
+DROP TABLE dml_test;
diff --git a/src/test/regress/failure_schedule b/src/test/regress/failure_schedule
index 2cefc7ba9..3642fbfbb 100644
--- a/src/test/regress/failure_schedule
+++ b/src/test/regress/failure_schedule
@@ -20,3 +20,4 @@ test: failure_1pc_copy_append
 test: failure_multi_shard_update_delete
 test: failure_cte_subquery
 test: failure_insert_select_via_coordinator
+test: failure_multi_dml
diff --git a/src/test/regress/sql/failure_multi_dml.sql b/src/test/regress/sql/failure_multi_dml.sql
new file mode 100644
index 000000000..15294c5b8
--- /dev/null
+++ b/src/test/regress/sql/failure_multi_dml.sql
@@ -0,0 +1,179 @@
+SELECT citus.mitmproxy('conn.allow()');
+
+SET citus.shard_count = 2;
+SET citus.shard_replication_factor = 1; -- one shard per worker
+SET citus.next_shard_id TO 103400;
+ALTER SEQUENCE pg_catalog.pg_dist_placement_placementid_seq RESTART 100;
+
+CREATE TABLE dml_test (id integer, name text);
+SELECT create_distributed_table('dml_test', 'id');
+
+COPY dml_test FROM STDIN WITH CSV;
+1,Alpha
+2,Beta
+3,Gamma
+4,Delta
+\.
+
+SELECT citus.clear_network_traffic();
+
+---- test multiple statements spanning multiple shards,
+---- at each significant point. These transactions are 2pc
+
+-- fail at DELETE
+SELECT citus.mitmproxy('conn.onQuery(query="^DELETE").kill()');
+
+BEGIN;
+DELETE FROM dml_test WHERE id = 1;
+DELETE FROM dml_test WHERE id = 2;
+INSERT INTO dml_test VALUES (5, 'Epsilon');
+UPDATE dml_test SET name = 'alpha' WHERE id = 1;
+UPDATE dml_test SET name = 'gamma' WHERE id = 3;
+COMMIT;
+
+--- shouldn't see any changes performed in failed transaction
+SELECT * FROM dml_test ORDER BY id ASC;
+
+-- fail at INSERT
+SELECT citus.mitmproxy('conn.onQuery(query="^INSERT").kill()');
+
+BEGIN;
+DELETE FROM dml_test WHERE id = 1;
+DELETE FROM dml_test WHERE id = 2;
+INSERT INTO dml_test VALUES (5, 'Epsilon');
+UPDATE dml_test SET name = 'alpha' WHERE id = 1;
+UPDATE dml_test SET name = 'gamma' WHERE id = 3;
+COMMIT;
+
+--- shouldn't see any changes before failed INSERT
+SELECT * FROM dml_test ORDER BY id ASC;
+
+-- fail at UPDATE
+SELECT citus.mitmproxy('conn.onQuery(query="^UPDATE").kill()');
+
+BEGIN;
+DELETE FROM dml_test WHERE id = 1;
+DELETE FROM dml_test WHERE id = 2;
+INSERT INTO dml_test VALUES (5, 'Epsilon');
+UPDATE dml_test SET name = 'alpha' WHERE id = 1;
+UPDATE dml_test SET name = 'gamma' WHERE id = 3;
+COMMIT;
+
+--- shouldn't see any changes after failed UPDATE
+SELECT * FROM dml_test ORDER BY id ASC;
+
+-- fail at PREPARE TRANSACTION
+SELECT citus.mitmproxy('conn.onQuery(query="^PREPARE TRANSACTION").kill()');
+
+-- hide the error message (it has the PID)...
+-- we'll test for the txn side-effects to ensure it didn't run
+SET client_min_messages TO FATAL;
+
+BEGIN;
+DELETE FROM dml_test WHERE id = 1;
+DELETE FROM dml_test WHERE id = 2;
+INSERT INTO dml_test VALUES (5, 'Epsilon');
+UPDATE dml_test SET name = 'alpha' WHERE id = 1;
+UPDATE dml_test SET name = 'gamma' WHERE id = 3;
+COMMIT;
+
+SET client_min_messages TO DEFAULT;
+
+SELECT citus.mitmproxy('conn.allow()');
+SELECT shardid FROM pg_dist_shard_placement WHERE shardstate = 3;
+SELECT recover_prepared_transactions();
+
+-- shouldn't see any changes after failed PREPARE
+SELECT * FROM dml_test ORDER BY id ASC;
+
+-- fail at COMMIT
+SELECT citus.mitmproxy('conn.onQuery(query="^COMMIT").kill()');
+
+-- hide the error message (it has the PID)...
+-- we'll test for the txn side-effects to ensure it didn't run
+SET client_min_messages TO FATAL;
+
+BEGIN;
+DELETE FROM dml_test WHERE id = 1;
+DELETE FROM dml_test WHERE id = 2;
+INSERT INTO dml_test VALUES (5, 'Epsilon');
+UPDATE dml_test SET name = 'alpha' WHERE id = 1;
+UPDATE dml_test SET name = 'gamma' WHERE id = 3;
+COMMIT;
+
+SET client_min_messages TO DEFAULT;
+
+SELECT citus.mitmproxy('conn.allow()');
+SELECT shardid FROM pg_dist_shard_placement WHERE shardstate = 3;
+SELECT recover_prepared_transactions();
+
+-- should see changes, because of txn recovery
+SELECT * FROM dml_test ORDER BY id ASC;
+
+-- drop table and recreate with different replication/sharding
+
+DROP TABLE dml_test;
+SET citus.shard_count = 1;
+SET citus.shard_replication_factor = 2; -- two placements
+
+CREATE TABLE dml_test (id integer, name text);
+SELECT create_distributed_table('dml_test', 'id');
+
+COPY dml_test FROM STDIN WITH CSV;
+1,Alpha
+2,Beta
+3,Gamma
+4,Delta
+\.
+
+---- test multiple statements against a single shard, but with two placements
+
+-- fail at COMMIT (actually COMMIT this time, as no 2pc in use)
+SELECT citus.mitmproxy('conn.onQuery(query="^COMMIT").kill()');
+
+BEGIN;
+DELETE FROM dml_test WHERE id = 1;
+DELETE FROM dml_test WHERE id = 2;
+INSERT INTO dml_test VALUES (5, 'Epsilon');
+UPDATE dml_test SET name = 'alpha' WHERE id = 1;
+UPDATE dml_test SET name = 'gamma' WHERE id = 3;
+COMMIT;
+
+--- should see all changes, but they only went to one placement (other is unhealthy)
+SELECT * FROM dml_test ORDER BY id ASC;
+SELECT shardid FROM pg_dist_shard_placement WHERE shardstate = 3;
+
+SELECT citus.mitmproxy('conn.allow()');
+
+-- drop table and recreate as reference table
+
+DROP TABLE dml_test;
+SET citus.shard_count = 2;
+SET citus.shard_replication_factor = 1;
+
+CREATE TABLE dml_test (id integer, name text);
+SELECT create_reference_table('dml_test');
+
+COPY dml_test FROM STDIN WITH CSV;
+1,Alpha
+2,Beta
+3,Gamma
+4,Delta
+\.
+
+-- fail at COMMIT (by failing to PREPARE)
+SELECT citus.mitmproxy('conn.onQuery(query="^PREPARE").kill()');
+
+BEGIN;
+DELETE FROM dml_test WHERE id = 1;
+DELETE FROM dml_test WHERE id = 2;
+INSERT INTO dml_test VALUES (5, 'Epsilon');
+UPDATE dml_test SET name = 'alpha' WHERE id = 1;
+UPDATE dml_test SET name = 'gamma' WHERE id = 3;
+COMMIT;
+
+--- shouldn't see any changes after failed COMMIT
+SELECT * FROM dml_test ORDER BY id ASC;
+
+SELECT citus.mitmproxy('conn.allow()');
+DROP TABLE dml_test;