diff --git a/engine/packages/universaldb/src/driver/postgres/chunks.rs b/engine/packages/universaldb/src/driver/postgres/chunks.rs
new file mode 100644
index 0000000000..2a7390414b
--- /dev/null
+++ b/engine/packages/universaldb/src/driver/postgres/chunks.rs
@@ -0,0 +1,102 @@
+use std::{
+	collections::HashMap,
+	time::{Duration, Instant},
+};
+
+/// A partially received commit request is dropped once no piece of it has arrived for this long. The
+/// follower gives up on an attempt after its request timeout and resends from the first piece, so
+/// anything older can never complete.
+pub const PENDING_CHUNK_MAX_IDLE: Duration = Duration::from_secs(30);
+
+/// One piece of a commit request that a follower split across several NATS messages.
+pub struct CommitChunk {
+	pub client_node_id: Vec<u8>,
+	pub client_seq: u64,
+	pub attempt: u32,
+	pub index: u32,
+	pub count: u32,
+	pub data: Vec<u8>,
+}
+
+struct PendingRequest {
+	attempt: u32,
+	count: u32,
+	next_index: u32,
+	bytes: Vec<u8>,
+	updated_at: Instant,
+}
+
+/// Reassembles chunked commit requests on the leader.
+///
+/// NATS delivers one publisher's messages on a subject in order, so pieces of an attempt arrive in
+/// sequence unless a reconnect loses some. Any gap abandons the attempt instead of waiting for the
+/// missing piece: the follower's request times out and it resends every piece under a new attempt.
+/// Owned by the single commit subscriber task.
+#[derive(Default)]
+pub struct ChunkAssembler {
+	pending: HashMap<(Vec<u8>, u64), PendingRequest>,
+}
+
+impl ChunkAssembler {
+	/// Accepts one piece and returns the complete encoded request once its last piece arrives.
+	pub fn push(&mut self, chunk: CommitChunk, now: Instant) -> Option<Vec<u8>> {
+		let key = (chunk.client_node_id, chunk.client_seq);
+
+		if chunk.index == 0 {
+			if let Some(existing) = self.pending.get(&key) {
+				if chunk.attempt < existing.attempt {
+					return None;
+				}
+			}
+			if chunk.count <= 1 {
+				self.pending.remove(&key);
+				return Some(chunk.data);
+			}
+			self.pending.insert(
+				key,
+				PendingRequest {
+					attempt: chunk.attempt,
+					count: chunk.count,
+					next_index: 1,
+					bytes: chunk.data,
+					updated_at: now,
+				},
+			);
+			return None;
+		}
+
+		let Some(pending) = self.pending.get_mut(&key) else {
+			return None;
+		};
+		if chunk.attempt < pending.attempt {
+			return None;
+		}
+		if chunk.attempt != pending.attempt
+			|| chunk.count != pending.count
+			|| chunk.index != pending.next_index
+		{
+			self.pending.remove(&key);
+			return None;
+		}
+
+		pending.bytes.extend_from_slice(&chunk.data);
+		pending.next_index += 1;
+		pending.updated_at = now;
+
+		if pending.next_index == pending.count {
+			return self.pending.remove(&key).map(|pending| pending.bytes);
+		}
+		None
+	}
+
+	/// Drops requests that stopped receiving pieces, so a follower that died mid-send does not pin its
+	/// partial request in memory.
+	pub fn evict_idle(&mut self, now: Instant) {
+		self.pending
+			.retain(|_, pending| now.duration_since(pending.updated_at) < PENDING_CHUNK_MAX_IDLE);
+	}
+}
+
+#[cfg(test)]
+#[path = "../../../tests/unit/postgres_chunks.rs"]
+mod tests;
diff --git a/engine/packages/universaldb/src/driver/postgres/codec.rs b/engine/packages/universaldb/src/driver/postgres/codec.rs
index 7e8fbda26b..415ff1b00f 100644
--- a/engine/packages/universaldb/src/driver/postgres/codec.rs
+++ b/engine/packages/universaldb/src/driver/postgres/codec.rs
@@ -1,4 +1,4 @@
-use anyhow::Result;
+use anyhow::{Context, Result};
 use rivet_universaldb_commit::{self as proto, versioned};
 use vbare::OwnedVersionedData;
 
@@ -7,7 +7,11 @@ use crate::{
 	tx_ops::Operation,
 };
 
-use super::transport::CommitOutcome;
+use super::{chunks::CommitChunk, transport::CommitOutcome};
+
+/// Protocol version that introduced [`proto::CommitRequestChunk`]. A fleet negotiated below it has
+/// leaders that cannot reassemble a chunked request.
+pub const CHUNKED_COMMIT_PROTOCOL_VERSION: u16 = 2;
 
 /// Decoded form of a commit request payload sent from a follower to the leader over NATS.
 pub struct DecodedCommit {
@@ -80,6 +84,60 @@ pub fn decode_commit_request(payload: &[u8]) -> Result<DecodedCommit> {
 	})
 }
 
+/// Split an encoded commit request into chunk messages that each fit within `max_payload` bytes,
+/// encoded at `protocol_version`, which must be at least [`CHUNKED_COMMIT_PROTOCOL_VERSION`].
+pub fn encode_commit_request_chunks(
+	request: &[u8],
+	client_node_id: &[u8],
+	client_seq: u64,
+	attempt: u32,
+	max_payload: usize,
+	protocol_version: u16,
+) -> Result<Vec<Vec<u8>>> {
+	let encode = |index: u32, count: u32, data: Vec<u8>| {
+		versioned::CommitRequestChunk::wrap_latest(proto::CommitRequestChunk {
+			client_node_id: client_node_id.to_vec(),
+			client_seq,
+			attempt,
+			index,
+			count,
+			data,
+		})
+		.serialize_with_embedded_version(protocol_version)
+	};
+
+	// Every field except `data` has the same encoded width in every chunk, so an empty chunk measures
+	// the envelope. The length prefix of `data` grows from one byte to at most five as a piece grows.
+	let overhead = encode(0, 0, Vec::new())?.len() + 4;
+	let piece_len = max_payload
+		.checked_sub(overhead)
+		.filter(|len| *len > 0)
+		.with_context(|| {
+			format!("nats max_payload of {max_payload} bytes cannot fit a commit request chunk")
+		})?;
+	let count = u32::try_from(request.len().div_ceil(piece_len))
+		.context("commit request needs too many chunks")?;
+
+	request
+		.chunks(piece_len)
+		.zip(0..)
+		.map(|(piece, index)| encode(index, count, piece.to_vec()))
+		.collect()
+}
+
+/// Decode one chunk produced by [`encode_commit_request_chunks`].
+pub fn decode_commit_request_chunk(payload: &[u8]) -> Result<CommitChunk> {
+	let chunk = versioned::CommitRequestChunk::deserialize_with_embedded_version(payload)?;
+	Ok(CommitChunk {
+		client_node_id: chunk.client_node_id,
+		client_seq: chunk.client_seq,
+		attempt: chunk.attempt,
+		index: chunk.index,
+		count: chunk.count,
+		data: chunk.data,
+	})
+}
+
 /// Encode a leader's commit reply at `protocol_version`, the version negotiated across the fleet,
 /// with an embedded version header.
 pub fn encode_commit_reply(outcome: CommitOutcome, protocol_version: u16) -> Result<Vec<u8>> {
diff --git a/engine/packages/universaldb/src/driver/postgres/commit.rs b/engine/packages/universaldb/src/driver/postgres/commit.rs
index 3310ca5b43..08e18d9996 100644
--- a/engine/packages/universaldb/src/driver/postgres/commit.rs
+++ b/engine/packages/universaldb/src/driver/postgres/commit.rs
@@ -3,7 +3,8 @@ use std::{
 	time::{Duration, Instant},
 };
 
-use anyhow::{Context, Result};
+use anyhow::{Context, Result, bail};
+use futures_util::FutureExt;
 use tokio::sync::oneshot;
 
 use crate::{
@@ -30,6 +31,8 @@ const REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
 const MAX_SUBMIT_ATTEMPTS: usize = 8;
 /// Backoff between multi-node resends.
 const RESEND_BACKOFF: Duration = Duration::from_millis(100);
+/// The NATS server's `max_payload` when it is not configured.
+const NATS_DEFAULT_MAX_PAYLOAD: usize = 1024 * 1024;
 
 /// Submit a follower transaction's commit to the leader and await the result.
 ///
@@ -106,22 +109,51 @@ async fn submit_nats(
 	// One dedup key for this logical commit, reused across every resend so the leader applies it at
 	// most once even if an earlier attempt was applied but its reply was lost to a failover.
 	let client_seq = shared.next_commit_seq();
+	let protocol_version = shared.commit_protocol_version();
 	let payload = codec::encode_commit_request(
 		read_version.max(0) as u64,
 		&conflict_ranges,
 		&operations,
 		shared.node_id.as_bytes(),
 		client_seq as u64,
-		shared.commit_protocol_version(),
+		protocol_version,
 	)
 	.context("failed to encode commit request")?;
 
 	let submit_start = Instant::now();
 	for attempt in 0..MAX_SUBMIT_ATTEMPTS {
 		let lease = wait_for_leader(shared).await?;
-		let subject = nats.subjects.commit(&lease.leader_addr);
 
-		let request = nats.client.request(subject, payload.clone().into());
+		// async-nats does not check a request against the server's max_payload, and the server
+		// answers an oversized message by closing the whole connection. A request that would not fit
+		// is split into chunks instead, which only a leader at the chunked protocol version accepts.
+		let max_payload = nats_max_payload(&nats.client);
+		let request = if payload.len() <= max_payload {
+			let subject = nats.subjects.commit(&lease.leader_addr);
+			let request = nats.client.request(subject, payload.clone().into());
+			async { request.await.map_err(anyhow::Error::from) }.boxed()
+		} else {
+			if protocol_version < codec::CHUNKED_COMMIT_PROTOCOL_VERSION {
+				bail!(
+					"commit request is {} bytes, over the nats server max_payload of {max_payload} bytes, \
+					 and the fleet has not negotiated chunked commit requests (protocol version \
+					 {protocol_version}); raise the nats max_payload or finish upgrading every node",
+					payload.len()
+				);
+			}
+			let chunks = codec::encode_commit_request_chunks(
+				&payload,
+				shared.node_id.as_bytes(),
+				client_seq as u64,
+				attempt as u32,
+				max_payload,
+				protocol_version,
+			)
+			.context("failed to chunk commit request")?;
+			let subject = nats.subjects.commit_chunk(&lease.leader_addr);
+			send_chunks(&nats.client, subject, chunks).boxed()
+		};
+
 		match tokio::time::timeout(REQUEST_TIMEOUT, request).await {
 			Ok(Ok(msg)) => match codec::decode_commit_reply(&msg.payload) {
 				Ok(CommitOutcome::Committed { .. }) => {
@@ -176,6 +208,38 @@ async fn submit_nats(
 	)
 }
 
+/// The largest message the connected NATS server accepts.
+fn nats_max_payload(client: &async_nats::Client) -> usize {
+	match client.server_info().max_payload {
+		// A client that has not received the server's INFO yet reports zero, so assume the server
+		// default rather than refusing to send.
+		0 => NATS_DEFAULT_MAX_PAYLOAD,
+		max_payload => max_payload,
+	}
+}
+
+/// Send every chunk except the last as a plain publish and the last as the request, so the reply
+/// arrives once the leader holds the whole commit.
+async fn send_chunks(
+	client: &async_nats::Client,
+	subject: String,
+	mut chunks: Vec<Vec<u8>>,
+) -> Result<async_nats::Message> {
+	let last = chunks
+		.pop()
+		.context("chunked commit request has no chunks")?;
+	for chunk in chunks {
+		client
+			.publish(subject.clone(), chunk.into())
+			.await
+			.context("failed to publish commit request chunk")?;
+	}
+	client
+		.request(subject, last.into())
+		.await
+		.context("commit request chunk got no reply")
+}
+
 /// Wait for a known leader, returning a retryable error if none is elected in time.
 async fn wait_for_leader(shared: &Arc<PostgresShared>) -> Result<LeaseInfo> {
 	let deadline = Instant::now() + LEADER_WAIT_TIMEOUT;
diff --git a/engine/packages/universaldb/src/driver/postgres/mod.rs b/engine/packages/universaldb/src/driver/postgres/mod.rs
index 20b80765c0..896e78bb8b 100644
--- a/engine/packages/universaldb/src/driver/postgres/mod.rs
+++ b/engine/packages/universaldb/src/driver/postgres/mod.rs
@@ -1,3 +1,4 @@
+mod chunks;
 mod codec;
 mod commit;
 mod database;
diff --git a/engine/packages/universaldb/src/driver/postgres/nats.rs b/engine/packages/universaldb/src/driver/postgres/nats.rs
index 0edcd2f0aa..ed12bb75e4 100644
--- a/engine/packages/universaldb/src/driver/postgres/nats.rs
+++ b/engine/packages/universaldb/src/driver/postgres/nats.rs
@@ -1,10 +1,11 @@
-use std::{str::FromStr, sync::Arc};
+use std::{str::FromStr, sync::Arc, time::Instant};
 
 use anyhow::{Context, Result};
 use futures_util::StreamExt;
-use tokio::sync::mpsc;
+use tokio::{sync::mpsc, time::MissedTickBehavior};
 
 use super::{
+	chunks::{ChunkAssembler, PENDING_CHUNK_MAX_IDLE},
 	codec,
 	shared::PostgresShared,
 	transport::{CommitJob, DedupKey, Responder},
@@ -49,6 +50,12 @@ impl Subjects {
 		format!("{}.commit.{leader_id}", self.prefix)
 	}
 
+	/// Subject a follower sends the pieces of a chunked commit request to. It is separate from
+	/// [`Subjects::commit`] so a leader that predates chunking never receives a piece it cannot decode.
+	pub fn commit_chunk(&self, leader_id: &str) -> String {
+		format!("{}.commit_chunk.{leader_id}", self.prefix)
+	}
+
 	/// Subject the leader publishes each watermark advance to; every node subscribes.
 	pub fn watermark(&self) -> String {
 		format!("{}.watermark", self.prefix)
@@ -86,58 +93,109 @@ pub async fn connect(config: &NatsConfig) -> Result<async_nats::Client> {
 		.context("failed to connect udb nats client")
 }
 
-/// Leader-side task: subscribe to this leader's commit subject, decode each request into a
-/// [`CommitJob`], and forward it into the drain loop's job queue. Returns when the subscription ends
-/// (client closed) or the drain loop's receiver is dropped (step-down).
+/// Leader-side task: subscribe to this leader's commit and commit chunk subjects, decode each request
+/// into a [`CommitJob`], and forward it into the drain loop's job queue. Returns when a subscription
+/// ends (client closed) or the drain loop's receiver is dropped (step-down).
 pub async fn run_commit_subscriber(
 	shared: &Arc<PostgresShared>,
 	client: async_nats::Client,
 	subject: String,
+	chunk_subject: String,
 	jobs_tx: mpsc::Sender<CommitJob>,
 ) -> Result<()> {
 	let mut sub = client
 		.subscribe(subject.clone())
 		.await
 		.with_context(|| format!("failed to subscribe to udb commit subject {subject}"))?;
-
-	while let Some(msg) = sub.next().await {
-		let Some(reply) = msg.reply.clone() else {
-			tracing::warn!("udb commit request missing reply subject; dropping");
-			continue;
-		};
-
-		let decoded = match codec::decode_commit_request(&msg.payload) {
-			Ok(decoded) => decoded,
-			Err(err) => {
-				tracing::warn!(?err, "failed to decode udb commit request; dropping");
-				continue;
+	let mut chunk_sub = client
+		.subscribe(chunk_subject.clone())
+		.await
+		.with_context(|| {
+			format!("failed to subscribe to udb commit chunk subject {chunk_subject}")
+		})?;
+
+	let mut assembler = ChunkAssembler::default();
+	let mut evict_interval = tokio::time::interval(PENDING_CHUNK_MAX_IDLE);
+	evict_interval.set_missed_tick_behavior(MissedTickBehavior::Delay);
+
+	loop {
+		tokio::select! {
+			msg = sub.next() => {
+				let Some(msg) = msg else {
+					break;
+				};
+				if !enqueue_request(shared, &client, msg.reply, &msg.payload, &jobs_tx).await {
+					break;
+				}
+			}
+			msg = chunk_sub.next() => {
+				let Some(msg) = msg else {
+					break;
+				};
+				let chunk = match codec::decode_commit_request_chunk(&msg.payload) {
+					Ok(chunk) => chunk,
+					Err(err) => {
+						tracing::warn!(?err, "failed to decode udb commit request chunk; dropping");
+						continue;
+					}
+				};
+				let Some(payload) = assembler.push(chunk, Instant::now()) else {
+					continue;
+				};
+				if !enqueue_request(shared, &client, msg.reply, &payload, &jobs_tx).await {
+					break;
+				}
+			}
+			_ = evict_interval.tick() => {
+				assembler.evict_idle(Instant::now());
 			}
-		};
-
-		let job = CommitJob {
-			read_version: decoded.read_version,
-			conflict_ranges: decoded.conflict_ranges,
-			operations: decoded.operations,
-			dedup_key: Some(DedupKey {
-				client_node_id: decoded.client_node_id,
-				client_seq: decoded.client_seq as i64,
-			}),
-			responder: Responder::Nats {
-				client: client.clone(),
-				reply,
-				protocol_version: shared.commit_protocol_version(),
-			},
-		};
-
-		// A full queue applies backpressure; a closed queue means the drain loop stepped down.
-		if jobs_tx.send(job).await.is_err() {
-			break;
 		}
 	}
 
 	Ok(())
 }
 
+/// Decode one complete commit request and queue it for the drain loop. Returns false once the drain
+/// loop has stepped down and no longer accepts jobs.
+async fn enqueue_request(
+	shared: &Arc<PostgresShared>,
+	client: &async_nats::Client,
+	reply: Option<async_nats::Subject>,
+	payload: &[u8],
+	jobs_tx: &mpsc::Sender<CommitJob>,
+) -> bool {
+	let Some(reply) = reply else {
+		tracing::warn!("udb commit request missing reply subject; dropping");
+		return true;
+	};
+
+	let decoded = match codec::decode_commit_request(payload) {
+		Ok(decoded) => decoded,
+		Err(err) => {
+			tracing::warn!(?err, "failed to decode udb commit request; dropping");
+			return true;
+		}
+	};
+
+	let job = CommitJob {
+		read_version: decoded.read_version,
+		conflict_ranges: decoded.conflict_ranges,
+		operations: decoded.operations,
+		dedup_key: Some(DedupKey {
+			client_node_id: decoded.client_node_id,
+			client_seq: decoded.client_seq as i64,
+		}),
+		responder: Responder::Nats {
+			client: client.clone(),
+			reply,
+			protocol_version: shared.commit_protocol_version(),
+		},
+	};
+
+	// A full queue applies backpressure; a closed queue means the drain loop stepped down.
+	jobs_tx.send(job).await.is_ok()
+}
+
 /// FNV-1a 64-bit hash. Deterministic across processes (unlike `DefaultHasher`), used only to derive a
 /// stable cluster subject prefix.
 fn fnv1a_64(bytes: &[u8]) -> u64 {
diff --git a/engine/packages/universaldb/src/driver/postgres/resolver/mod.rs b/engine/packages/universaldb/src/driver/postgres/resolver/mod.rs
index eaff651ef2..db222fb119 100644
--- a/engine/packages/universaldb/src/driver/postgres/resolver/mod.rs
+++ b/engine/packages/universaldb/src/driver/postgres/resolver/mod.rs
@@ -187,9 +187,12 @@ fn spawn_commit_subscriber(
 	};
 	let client = nats.client.clone();
 	let subject = nats.subjects.commit(&shared.node_id);
+	let chunk_subject = nats.subjects.commit_chunk(&shared.node_id);
 	let shared = shared.clone();
 	Some(AbortOnDropHandle::new(tokio::spawn(async move {
-		if let Err(err) = super::nats::run_commit_subscriber(&shared, client, subject, tx).await {
+		if let Err(err) =
+			super::nats::run_commit_subscriber(&shared, client, subject, chunk_subject, tx).await
+		{
 			tracing::warn!(?err, "udb commit subscriber ended");
 		}
 	})))
diff --git a/engine/packages/universaldb/tests/nats_large_commit.rs b/engine/packages/universaldb/tests/nats_large_commit.rs
new file mode 100644
index 0000000000..ce52b345c7
--- /dev/null
+++ b/engine/packages/universaldb/tests/nats_large_commit.rs
@@ -0,0 +1,326 @@
+//! Follower commits whose encoded request is larger than the NATS server's `max_payload`.
+//!
+//! A follower sends each commit to the leader as one NATS request, and NATS rejects any message over
+//! the server's `max_payload`, which defaults to 1 MiB. A transaction that writes more than that must
+//! still commit, and must not disturb the follower's other commits while it does.
+
+use std::{
+	process::Command,
+	sync::Arc,
+	time::{Duration, Instant},
+};
+
+use rivet_test_deps_docker::{TestDatabase, TestPubSub};
+use tokio::sync::watch;
+use tokio_postgres::NoTls;
+use universaldb::{
+	Database,
+	driver::postgres::{NatsConfig, PostgresConfig},
+	utils::IsolationLevel::*,
+};
+use uuid::Uuid;
+
+/// Enough 64 KiB values to push the encoded commit request well past the 1 MiB NATS default.
+const LARGE_VALUE_BYTES: usize = 64 * 1024;
+const LARGE_VALUE_COUNT: usize = 24;
+
+/// Chunked commit requests arrived in this commit protocol version.
+const CHUNKED_COMMIT_PROTOCOL_VERSION: u16 = 2;
+
+/// A config whose fleet has negotiated `commit_protocol_version`.
+fn test_config(commit_protocol_version: u16) -> rivet_config::Config {
+	rivet_config::Config::from_root_with_build_meta(
+		rivet_config::config::Root::default(),
+		rivet_config::BuildMeta::default(),
+		rivet_config::RuntimeProtocols {
+			universaldb_commit: rivet_config::RuntimeProtocol::new(
+				rivet_config::RuntimeProtocolKind::UniversaldbCommit,
+				commit_protocol_version,
+			),
+			..Default::default()
+		},
+	)
+}
+
+async fn setup_postgres() -> (String, rivet_test_deps_docker::DockerRunConfig) {
+	let (db_config, docker_config) = TestDatabase::Postgres
+		.config(Uuid::new_v4(), 1)
+		.await
+		.unwrap();
+	let mut docker_config = docker_config.unwrap();
+	docker_config.start().await.unwrap();
+	TestDatabase::Postgres
+		.wait_for_ready(&docker_config)
+		.await
+		.unwrap();
+	let rivet_config::config::Database::Postgres(postgres_config) = db_config else {
+		unreachable!();
+	};
+	(postgres_config.url.read().clone(), docker_config)
+}
+
+/// The test image runs NATS with its default configuration, so `max_payload` is 1 MiB.
+async fn setup_nats() -> (NatsConfig, rivet_test_deps_docker::DockerRunConfig) {
+	let (pubsub_config, docker_config) = TestPubSub::Nats.config(Uuid::new_v4(), 1).await.unwrap();
+	let mut docker_config = docker_config.unwrap();
+	docker_config.start().await.unwrap();
+	tokio::time::sleep(Duration::from_secs(1)).await;
+	let rivet_config::config::PubSub::Nats(nats) = pubsub_config else {
+		unreachable!();
+	};
+	(
+		NatsConfig {
+			addresses: nats.addresses.clone(),
+			username: nats.username.clone(),
+			password: nats.password.as_ref().map(|p| p.read().clone()),
+			client_capacity: nats.client_capacity,
+			subscription_capacity: nats.subscription_capacity,
+		},
+		docker_config,
+	)
+}
+
+async fn make_db(
+	connection_string: &str,
+	nats: &NatsConfig,
+	commit_protocol_version: u16,
+) -> Database {
+	let mut config = PostgresConfig::new(connection_string.to_string());
+	config.nats = Some(nats.clone());
+	let driver = universaldb::driver::PostgresDatabaseDriver::new_with_config(
+		test_config(commit_protocol_version),
+		config,
+	)
+	.await
+	.unwrap();
+	Database::new(Arc::new(driver))
+}
+
+async fn write_large(db: &Database) -> anyhow::Result<()> {
+	db.txn("test_large_commit", |tx| async move {
+		for i in 0..LARGE_VALUE_COUNT {
+			tx.set(&large_key(i), &vec![b'x'; LARGE_VALUE_BYTES]);
+		}
+		Ok(())
+	})
+	.await
+}
+
+/// Waits for the first node's election, so the node created next runs as a follower.
+async fn wait_for_leader(connection_string: &str) {
+	let (client, connection) = tokio_postgres::connect(connection_string, NoTls)
+		.await
+		.unwrap();
+	tokio::spawn(async move {
+		let _ = connection.await;
+	});
+	let deadline = Instant::now() + Duration::from_secs(15);
+	loop {
+		// The lease row is written by another process, so there is no event to await here.
+		let elected = client
+			.query_opt("SELECT epoch FROM udb_lease WHERE id = 1", &[])
+			.await
+			.unwrap()
+			.is_some();
+		if elected {
+			return;
+		}
+		assert!(Instant::now() < deadline, "no leader elected");
+		tokio::time::sleep(Duration::from_millis(100)).await;
+	}
+}
+
+fn small_key(i: u32) -> Vec<u8> {
+	format!("nats_large_commit/small/{i:08}").into_bytes()
+}
+
+fn large_key(i: usize) -> Vec<u8> {
+	format!("nats_large_commit/large/{i:04}").into_bytes()
+}
+
+async fn write_small(db: &Database, i: u32) -> anyhow::Result<()> {
+	db.txn("test_small_commit", move |tx| async move {
+		tx.set(&small_key(i), b"v");
+		Ok(())
+	})
+	.await
+}
+
+/// Lines of the NATS server log that report a message over `max_payload`.
+fn nats_payload_violations(container_name: &str) -> Vec<String> {
+	let output = Command::new("docker")
+		.args(["logs", container_name])
+		.output()
+		.unwrap();
+	let mut logs = String::from_utf8_lossy(&output.stdout).into_owned();
+	logs.push_str(&String::from_utf8_lossy(&output.stderr));
+	logs.lines()
+		.filter(|line| line.to_lowercase().contains("payload"))
+		.map(str::to_string)
+		.collect()
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn follower_commit_larger_than_nats_max_payload() {
+	let _ = tracing_subscriber::fmt()
+		.with_env_filter("warn")
+		.with_test_writer()
+		.try_init();
+
+	let (connection_string, _postgres_docker) = setup_postgres().await;
+	let (nats, nats_docker) = setup_nats().await;
+
+	let leader = make_db(
+		&connection_string,
+		&nats,
+		rivet_universaldb_commit::PROTOCOL_VERSION,
+	)
+	.await;
+	wait_for_leader(&connection_string).await;
+	let follower = Arc::new(
+		make_db(
+			&connection_string,
+			&nats,
+			rivet_universaldb_commit::PROTOCOL_VERSION,
+		)
+		.await,
+	);
+
+	write_small(&follower, 0)
+		.await
+		.expect("follower commits normally before the large commit");
+
+	// Small commits keep flowing from the same follower for as long as the large commit is in flight,
+	// so any disruption the large request causes to the shared NATS connection shows up here.
+	let (stop_tx, mut stop_rx) = watch::channel(false);
+	let small_writer = tokio::spawn({
+		let follower = follower.clone();
+		async move {
+			let mut slowest = Duration::ZERO;
+			let mut failures = Vec::new();
+			let mut completed = 0u32;
+			loop {
+				let start = Instant::now();
+				tokio::select! {
+					res = write_small(&follower, completed + 1) => {
+						slowest = slowest.max(start.elapsed());
+						if let Err(err) = res {
+							failures.push(format!("{err:#}"));
+						}
+						completed += 1;
+					}
+					_ = stop_rx.changed() => break,
+				}
+			}
+			(completed, slowest, failures)
+		}
+	});
+
+	let large_start = Instant::now();
+	let large = tokio::time::timeout(
+		Duration::from_secs(120),
+		follower.txn("test_large_commit", |tx| async move {
+			// One attempt is enough to show whether the request can be delivered at all.
+			tx.retry_limit(0)?;
+			for i in 0..LARGE_VALUE_COUNT {
+				tx.set(&large_key(i), &vec![b'x'; LARGE_VALUE_BYTES]);
+			}
+			Ok(())
+		}),
+	)
+	.await;
+	let large_elapsed = large_start.elapsed();
+
+	stop_tx.send(true).unwrap();
+	let (small_completed, small_slowest, small_failures) = small_writer.await.unwrap();
+	let violations = nats_payload_violations(&nats_docker.container_name);
+
+	let large_outcome = match &large {
+		Ok(Ok(())) => "committed".to_string(),
+		Ok(Err(err)) => format!("failed: {err:#}"),
+		Err(_) => "timed out".to_string(),
+	};
+	println!(
+		"large commit: {large_outcome} after {large_elapsed:?}\n\
+		 small commits during it: completed={small_completed} slowest={small_slowest:?} failures={small_failures:#?}\n\
+		 nats payload violations: {violations:#?}"
+	);
+
+	assert!(
+		matches!(large, Ok(Ok(()))),
+		"large follower commit did not commit: {large_outcome}"
+	);
+	assert!(
+		violations.is_empty(),
+		"a commit request exceeded the NATS max_payload: {violations:#?}"
+	);
+	assert!(
+		small_failures.is_empty(),
+		"small commits failed while the large commit was in flight: {small_failures:#?}"
+	);
+
+	let stored = leader
+		.txn("test_read_large_commit", |tx| async move {
+			let mut total = 0usize;
+			for i in 0..LARGE_VALUE_COUNT {
+				if let Some(value) = tx.get(&large_key(i), Serializable).await? {
+					total += value.len();
+				}
+			}
+			Ok(total)
+		})
+		.await
+		.unwrap();
+	assert_eq!(stored, LARGE_VALUE_BYTES * LARGE_VALUE_COUNT);
+}
+
+/// Until every node understands chunked commit requests, an oversized commit must fail right away
+/// with an error naming the limit, rather than sending a message the NATS server answers by closing
+/// the follower's connection.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn oversized_commit_fails_fast_before_chunking_is_negotiated() {
+	let _ = tracing_subscriber::fmt()
+		.with_env_filter("warn")
+		.with_test_writer()
+		.try_init();
+
+	let (connection_string, _postgres_docker) = setup_postgres().await;
+	let (nats, nats_docker) = setup_nats().await;
+
+	let _leader = make_db(
+		&connection_string,
+		&nats,
+		CHUNKED_COMMIT_PROTOCOL_VERSION - 1,
+	)
+	.await;
+	wait_for_leader(&connection_string).await;
+	let follower = make_db(
+		&connection_string,
+		&nats,
+		CHUNKED_COMMIT_PROTOCOL_VERSION - 1,
+	)
+	.await;
+
+	// No retry limit is set, so a retryable failure would keep resending well past one request
+	// timeout and fail the elapsed check below.
+	let start = Instant::now();
+	let err = write_large(&follower)
+		.await
+		.expect_err("an oversized commit cannot be delivered before chunking is negotiated");
+	let elapsed = start.elapsed();
+
+	let message = format!("{err:#}");
+	assert!(
+		message.contains("max_payload"),
+		"error should name the nats limit: {message}"
+	);
+	assert!(
+		elapsed < Duration::from_secs(5),
+		"oversized commit took {elapsed:?} to fail"
+	);
+	let violations = nats_payload_violations(&nats_docker.container_name);
+	assert!(
+		violations.is_empty(),
+		"the oversized request reached the nats server: {violations:#?}"
+	);
+}
diff --git a/engine/packages/universaldb/tests/unit/postgres_chunks.rs b/engine/packages/universaldb/tests/unit/postgres_chunks.rs
new file mode 100644
index 0000000000..e32cad1ac1
--- /dev/null
+++ b/engine/packages/universaldb/tests/unit/postgres_chunks.rs
@@ -0,0 +1,131 @@
+//! Chunked commit request encoding and leader-side reassembly.
+
+use std::time::Instant;
+
+use super::{
+	super::codec::{self, CHUNKED_COMMIT_PROTOCOL_VERSION},
+	ChunkAssembler, CommitChunk, PENDING_CHUNK_MAX_IDLE,
+};
+
+const NODE_ID: &[u8] = b"follower-node";
+const CLIENT_SEQ: u64 = 7;
+
+fn chunk(attempt: u32, index: u32, count: u32, data: &[u8]) -> CommitChunk {
+	CommitChunk {
+		client_node_id: NODE_ID.to_vec(),
+		client_seq: CLIENT_SEQ,
+		attempt,
+		index,
+		count,
+		data: data.to_vec(),
+	}
+}
+
+#[test]
+fn encoded_chunks_fit_max_payload_and_reassemble() {
+	let request: Vec<u8> = (0..300_000u32).map(|i| i as u8).collect();
+
+	for max_payload in [256, 4096, 65_536, 1024 * 1024] {
+		let encoded = codec::encode_commit_request_chunks(
+			&request,
+			NODE_ID,
+			CLIENT_SEQ,
+			3,
+			max_payload,
+			CHUNKED_COMMIT_PROTOCOL_VERSION,
+		)
+		.unwrap();
+		assert!(
+			encoded.iter().all(|bytes| bytes.len() <= max_payload),
+			"a chunk exceeded max_payload {max_payload}"
+		);
+
+		let mut assembler = ChunkAssembler::default();
+		let now = Instant::now();
+		let last = encoded.len() - 1;
+		for (i, bytes) in encoded.iter().enumerate() {
+			let decoded = codec::decode_commit_request_chunk(bytes).unwrap();
+			let assembled = assembler.push(decoded, now);
+			if i < last {
+				assert!(
+					assembled.is_none(),
+					"request completed before its last chunk"
+				);
+			} else {
+				assert_eq!(assembled.as_deref(), Some(request.as_slice()));
+			}
+		}
+	}
+}
+
+#[test]
+fn chunks_require_the_chunked_protocol_version() {
+	let result = codec::encode_commit_request_chunks(
+		&[0; 1024],
+		NODE_ID,
+		CLIENT_SEQ,
+		0,
+		256,
+		CHUNKED_COMMIT_PROTOCOL_VERSION - 1,
+	);
+	assert!(result.is_err());
+}
+
+#[test]
+fn max_payload_smaller_than_the_envelope_is_rejected() {
+	let result = codec::encode_commit_request_chunks(
+		&[0; 1024],
+		NODE_ID,
+		CLIENT_SEQ,
+		0,
+		16,
+		CHUNKED_COMMIT_PROTOCOL_VERSION,
+	);
+	assert!(result.is_err());
+}
+
+#[test]
+fn gap_abandons_the_attempt_until_it_is_resent() {
+	let mut assembler = ChunkAssembler::default();
+	let now = Instant::now();
+
+	assert!(assembler.push(chunk(0, 0, 3, b"a"), now).is_none());
+	assert!(assembler.push(chunk(0, 2, 3, b"c"), now).is_none());
+	// The gap dropped the attempt, so the late middle piece cannot complete it.
+	assert!(assembler.push(chunk(0, 1, 3, b"b"), now).is_none());
+
+	assert!(assembler.push(chunk(1, 0, 3, b"a"), now).is_none());
+	assert!(assembler.push(chunk(1, 1, 3, b"b"), now).is_none());
+	assert_eq!(
+		assembler.push(chunk(1, 2, 3, b"c"), now).as_deref(),
+		Some(&b"abc"[..])
+	);
+}
+
+#[test]
+fn pieces_of_an_earlier_attempt_are_ignored() {
+	let mut assembler = ChunkAssembler::default();
+	let now = Instant::now();
+
+	assert!(assembler.push(chunk(1, 0, 2, b"new-"), now).is_none());
+	assert!(assembler.push(chunk(0, 1, 2, b"old"), now).is_none());
+	assert!(assembler.push(chunk(0, 0, 2, b"old-"), now).is_none());
+	assert_eq!(
+		assembler.push(chunk(1, 1, 2, b"piece"), now).as_deref(),
+		Some(&b"new-piece"[..])
+	);
+}
+
+#[test]
+fn idle_requests_are_evicted() {
+	let mut assembler = ChunkAssembler::default();
+	let start = Instant::now();
+
+	assert!(assembler.push(chunk(0, 0, 2, b"a"), start).is_none());
+	assembler.evict_idle(start + PENDING_CHUNK_MAX_IDLE);
+	assert!(
+		assembler
+			.push(chunk(0, 1, 2, b"b"), start + PENDING_CHUNK_MAX_IDLE)
+			.is_none()
+	);
+}
diff --git a/engine/sdks/rust/universaldb-commit/src/lib.rs b/engine/sdks/rust/universaldb-commit/src/lib.rs
index 41954bbe75..0cb85fc13a 100644
--- a/engine/sdks/rust/universaldb-commit/src/lib.rs
+++ b/engine/sdks/rust/universaldb-commit/src/lib.rs
@@ -3,4 +3,4 @@ pub mod versioned;
 
 // Re-export latest
 pub use generated::PROTOCOL_VERSION;
-pub use generated::v1::*;
+pub use generated::v2::*;
diff --git a/engine/sdks/rust/universaldb-commit/src/versioned.rs b/engine/sdks/rust/universaldb-commit/src/versioned.rs
deleted file mode 100644
index 804f533a64..0000000000
--- a/engine/sdks/rust/universaldb-commit/src/versioned.rs
+++ /dev/null
@@ -1,100 +0,0 @@
-use anyhow::{Ok, Result, bail};
-use vbare::OwnedVersionedData;
-
-use crate::generated::v1;
-
-// Only v1 exists today. When adding v2+, generate converters with
-// `scripts/vbare-gen-converters` (see the envoy-protocol package for the
-// resulting `versioned/` module layout) and wire them in here.
-pub enum CommitRequest {
-	V1(v1::CommitRequest),
-}
-
-impl OwnedVersionedData for CommitRequest {
-	type Latest = v1::CommitRequest;
-
-	fn wrap_latest(latest: v1::CommitRequest) -> Self {
-		CommitRequest::V1(latest)
-	}
-
-	fn unwrap_latest(self) -> Result<Self::Latest> {
-		match self {
-			CommitRequest::V1(data) => Ok(data),
-		}
-	}
-
-	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
-		match version {
-			1 => Ok(CommitRequest::V1(serde_bare::from_slice(payload)?)),
-			_ => bail!("invalid version: {version}"),
-		}
-	}
-
-	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
-		match self {
-			CommitRequest::V1(data) => serde_bare::to_vec(&data).map_err(Into::into),
-		}
-	}
-}
-
-pub enum CommitReply {
-	V1(v1::CommitReply),
-}
-
-impl OwnedVersionedData for CommitReply {
-	type Latest = v1::CommitReply;
-
-	fn wrap_latest(latest: v1::CommitReply) -> Self {
-		CommitReply::V1(latest)
-	}
-
-	fn unwrap_latest(self) -> Result<Self::Latest> {
-		match self {
-			CommitReply::V1(data) => Ok(data),
-		}
-	}
-
-	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
-		match version {
-			1 => Ok(CommitReply::V1(serde_bare::from_slice(payload)?)),
-			_ => bail!("invalid version: {version}"),
-		}
-	}
-
-	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
-		match self {
-			CommitReply::V1(data) => serde_bare::to_vec(&data).map_err(Into::into),
-		}
-	}
-}
-
-pub enum Watermark {
-	V1(v1::Watermark),
-}
-
-impl OwnedVersionedData for Watermark {
-	type Latest = v1::Watermark;
-
-	fn wrap_latest(latest: v1::Watermark) -> Self {
-		Watermark::V1(latest)
-	}
-
-	fn unwrap_latest(self) -> Result<Self::Latest> {
-		match self {
-			Watermark::V1(data) => Ok(data),
-		}
-	}
-
-	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
-		match version {
-			1 => Ok(Watermark::V1(serde_bare::from_slice(payload)?)),
-			_ => bail!("invalid version: {version}"),
-		}
-	}
-
-	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
-		match self {
-			Watermark::V1(data) => serde_bare::to_vec(&data).map_err(Into::into),
-		}
-	}
-}
diff --git a/engine/sdks/rust/universaldb-commit/src/versioned/mod.rs b/engine/sdks/rust/universaldb-commit/src/versioned/mod.rs
new file mode 100644
index 0000000000..59dc2a3358
--- /dev/null
+++ b/engine/sdks/rust/universaldb-commit/src/versioned/mod.rs
@@ -0,0 +1,236 @@
+use anyhow::{Ok, Result, bail};
+use vbare::OwnedVersionedData;
+
+use crate::generated::{v1, v2};
+
+mod v1_to_v2;
+mod v2_to_v1;
+
+pub enum CommitRequest {
+	V1(v1::CommitRequest),
+	V2(v2::CommitRequest),
+}
+
+impl OwnedVersionedData for CommitRequest {
+	type Latest = v2::CommitRequest;
+
+	fn wrap_latest(latest: v2::CommitRequest) -> Self {
+		CommitRequest::V2(latest)
+	}
+
+	fn unwrap_latest(self) -> Result<Self::Latest> {
+		match self {
+			CommitRequest::V2(data) => Ok(data),
+			CommitRequest::V1(_) => bail!("version not latest"),
+		}
+	}
+
+	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
+		match version {
+			1 => Ok(CommitRequest::V1(serde_bare::from_slice(payload)?)),
+			2 => Ok(CommitRequest::V2(serde_bare::from_slice(payload)?)),
+			_ => bail!("invalid version: {version}"),
+		}
+	}
+
+	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
+		match self {
+			CommitRequest::V1(data) => serde_bare::to_vec(&data).map_err(Into::into),
+			CommitRequest::V2(data) => serde_bare::to_vec(&data).map_err(Into::into),
+		}
+	}
+
+	fn deserialize_converters() -> Vec<impl Fn(Self) -> Result<Self>> {
+		vec![Self::v1_to_v2]
+	}
+
+	fn serialize_converters() -> Vec<impl Fn(Self) -> Result<Self>> {
+		vec![Self::v2_to_v1]
+	}
+}
+
+impl CommitRequest {
+	fn v1_to_v2(self) -> Result<Self> {
+		match self {
+			CommitRequest::V1(data) => Ok(CommitRequest::V2(
+				v1_to_v2::convert_commit_request_v1_to_v2(data)?,
+			)),
+			CommitRequest::V2(_) => bail!("unexpected version"),
+		}
+	}
+
+	fn v2_to_v1(self) -> Result<Self> {
+		match self {
+			CommitRequest::V2(data) => Ok(CommitRequest::V1(
+				v2_to_v1::convert_commit_request_v2_to_v1(data)?,
+			)),
+			CommitRequest::V1(_) => bail!("unexpected version"),
+		}
+	}
+}
+
+/// Chunked commit requests were introduced in v2, so there is no earlier version to convert from.
+/// The identity converters make vbare count v2 as this type's latest version.
+pub enum CommitRequestChunk {
+	V2(v2::CommitRequestChunk),
+}
+
+impl OwnedVersionedData for CommitRequestChunk {
+	type Latest = v2::CommitRequestChunk;
+
+	fn wrap_latest(latest: v2::CommitRequestChunk) -> Self {
+		CommitRequestChunk::V2(latest)
+	}
+
+	fn unwrap_latest(self) -> Result<Self::Latest> {
+		match self {
+			CommitRequestChunk::V2(data) => Ok(data),
+		}
+	}
+
+	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
+		match version {
+			2 => Ok(CommitRequestChunk::V2(serde_bare::from_slice(payload)?)),
+			_ => bail!("commit request chunks do not exist at version {version}"),
+		}
+	}
+
+	fn serialize_version(self, version: u16) -> Result<Vec<u8>> {
+		match (self, version) {
+			(CommitRequestChunk::V2(data), 2) => serde_bare::to_vec(&data).map_err(Into::into),
+			(CommitRequestChunk::V2(_), _) => {
+				bail!("commit request chunks do not exist at version {version}")
+			}
+		}
+	}
+
+	fn deserialize_converters() -> Vec<impl Fn(Self) -> Result<Self>> {
+		vec![Ok]
+	}
+
+	fn serialize_converters() -> Vec<impl Fn(Self) -> Result<Self>> {
+		vec![Ok]
+	}
+}
+
+pub enum CommitReply {
+	V1(v1::CommitReply),
+	V2(v2::CommitReply),
+}
+
+impl OwnedVersionedData for CommitReply {
+	type Latest = v2::CommitReply;
+
+	fn wrap_latest(latest: v2::CommitReply) -> Self {
+		CommitReply::V2(latest)
+	}
+
+	fn unwrap_latest(self) -> Result<Self::Latest> {
+		match self {
+			CommitReply::V2(data) => Ok(data),
+			CommitReply::V1(_) => bail!("version not latest"),
+		}
+	}
+
+	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
+		match version {
+			1 => Ok(CommitReply::V1(serde_bare::from_slice(payload)?)),
+			2 => Ok(CommitReply::V2(serde_bare::from_slice(payload)?)),
+			_ => bail!("invalid version: {version}"),
+		}
+	}
+
+	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
+		match self {
+			CommitReply::V1(data) => serde_bare::to_vec(&data).map_err(Into::into),
+			CommitReply::V2(data) => serde_bare::to_vec(&data).map_err(Into::into),
+		}
+	}
+
+	fn deserialize_converters() -> Vec<impl Fn(Self) -> Result<Self>> {
+		vec![Self::v1_to_v2]
+	}
+
+	fn serialize_converters() -> Vec<impl Fn(Self) -> Result<Self>> {
+		vec![Self::v2_to_v1]
+	}
+}
+
+impl CommitReply {
+	fn v1_to_v2(self) -> Result<Self> {
+		match self {
+			CommitReply::V1(data) => Ok(CommitReply::V2(v1_to_v2::convert_commit_reply_v1_to_v2(
+				data,
+			)?)),
+			CommitReply::V2(_) => bail!("unexpected version"),
+		}
+	}
+
+	fn v2_to_v1(self) -> Result<Self> {
+		match self {
+			CommitReply::V2(data) => Ok(CommitReply::V1(v2_to_v1::convert_commit_reply_v2_to_v1(
+				data,
+			)?)),
+			CommitReply::V1(_) => bail!("unexpected version"),
+		}
+	}
+}
+
+pub enum Watermark {
+	V1(v1::Watermark),
+	V2(v2::Watermark),
+}
+
+impl OwnedVersionedData for Watermark {
+	type Latest = v2::Watermark;
+
+	fn wrap_latest(latest: v2::Watermark) -> Self {
+		Watermark::V2(latest)
+	}
+
+	fn unwrap_latest(self) -> Result<Self::Latest> {
+		match self {
+			Watermark::V2(data) => Ok(data),
+			Watermark::V1(_) => bail!("version not latest"),
+		}
+	}
+
+	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
+		match version {
+			1 => Ok(Watermark::V1(serde_bare::from_slice(payload)?)),
+			2 => Ok(Watermark::V2(serde_bare::from_slice(payload)?)),
+			_ => bail!("invalid version: {version}"),
+		}
+	}
+
+	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
+		match self {
+			Watermark::V1(data) => serde_bare::to_vec(&data).map_err(Into::into),
+			Watermark::V2(data) => serde_bare::to_vec(&data).map_err(Into::into),
+		}
+	}
+
+	fn deserialize_converters() -> Vec<impl Fn(Self) -> Result<Self>> {
+		vec![Self::v1_to_v2]
+	}
+
+	fn serialize_converters() -> Vec<impl Fn(Self) -> Result<Self>> {
+		vec![Self::v2_to_v1]
+	}
+}
+
+impl Watermark {
+	fn v1_to_v2(self) -> Result<Self> {
+		match self {
+			Watermark::V1(data) => Ok(Watermark::V2(v1_to_v2::convert_watermark_v1_to_v2(data)?)),
+			Watermark::V2(_) => bail!("unexpected version"),
+		}
+	}
+
+	fn v2_to_v1(self) -> Result<Self> {
+		match self {
+			Watermark::V2(data) => Ok(Watermark::V1(v2_to_v1::convert_watermark_v2_to_v1(data)?)),
+			Watermark::V1(_) => bail!("unexpected version"),
+		}
+	}
+}
diff --git a/engine/sdks/rust/universaldb-commit/src/versioned/v1_to_v2.rs b/engine/sdks/rust/universaldb-commit/src/versioned/v1_to_v2.rs
new file mode 100644
index 0000000000..5591b973f1
--- /dev/null
+++ b/engine/sdks/rust/universaldb-commit/src/versioned/v1_to_v2.rs
@@ -0,0 +1,116 @@
+#![allow(dead_code, unused_variables)]
+
+use anyhow::Result;
+
+use crate::generated::{v1, v2};
+
+pub fn convert_conflict_range_type_v1_to_v2(
+	x: v1::ConflictRangeType,
+) -> Result<v2::ConflictRangeType> {
+	Ok(match x {
+		v1::ConflictRangeType::Read => v2::ConflictRangeType::Read,
+		v1::ConflictRangeType::Write => v2::ConflictRangeType::Write,
+	})
+}
+
+pub fn convert_conflict_range_v1_to_v2(x: v1::ConflictRange) -> Result<v2::ConflictRange> {
+	Ok(v2::ConflictRange {
+		begin: x.begin,
+		end: x.end,
+		kind: convert_conflict_range_type_v1_to_v2(x.kind)?,
+	})
+}
+
+pub fn convert_mutation_type_v1_to_v2(x: v1::MutationType) -> Result<v2::MutationType> {
+	Ok(match x {
+		v1::MutationType::Add => v2::MutationType::Add,
+		v1::MutationType::And => v2::MutationType::And,
+		v1::MutationType::BitAnd => v2::MutationType::BitAnd,
+		v1::MutationType::Or => v2::MutationType::Or,
+		v1::MutationType::BitOr => v2::MutationType::BitOr,
+		v1::MutationType::Xor => v2::MutationType::Xor,
+		v1::MutationType::BitXor => v2::MutationType::BitXor,
+		v1::MutationType::AppendIfFits => v2::MutationType::AppendIfFits,
+		v1::MutationType::Max => v2::MutationType::Max,
+		v1::MutationType::Min => v2::MutationType::Min,
+		v1::MutationType::SetVersionstampedKey => v2::MutationType::SetVersionstampedKey,
+		v1::MutationType::SetVersionstampedValue => v2::MutationType::SetVersionstampedValue,
+		v1::MutationType::ByteMin => v2::MutationType::ByteMin,
+		v1::MutationType::ByteMax => v2::MutationType::ByteMax,
+		v1::MutationType::CompareAndClear => v2::MutationType::CompareAndClear,
+	})
+}
+
+pub fn convert_set_value_v1_to_v2(x: v1::SetValue) -> Result<v2::SetValue> {
+	Ok(v2::SetValue {
+		key: x.key,
+		value: x.value,
+	})
+}
+
+pub fn convert_clear_v1_to_v2(x: v1::Clear) -> Result<v2::Clear> {
+	Ok(v2::Clear { key: x.key })
+}
+
+pub fn convert_clear_range_v1_to_v2(x: v1::ClearRange) -> Result<v2::ClearRange> {
+	Ok(v2::ClearRange {
+		begin: x.begin,
+		end: x.end,
+	})
+}
+
+pub fn convert_atomic_op_v1_to_v2(x: v1::AtomicOp) -> Result<v2::AtomicOp> {
+	Ok(v2::AtomicOp {
+		key: x.key,
+		param: x.param,
+		op_type: convert_mutation_type_v1_to_v2(x.op_type)?,
+	})
+}
+
+pub fn convert_operation_v1_to_v2(x: v1::Operation) -> Result<v2::Operation> {
+	Ok(match x {
+		v1::Operation::SetValue(v) => v2::Operation::SetValue(convert_set_value_v1_to_v2(v)?),
+		v1::Operation::Clear(v) => v2::Operation::Clear(convert_clear_v1_to_v2(v)?),
+		v1::Operation::ClearRange(v) => v2::Operation::ClearRange(convert_clear_range_v1_to_v2(v)?),
+		v1::Operation::AtomicOp(v) => v2::Operation::AtomicOp(convert_atomic_op_v1_to_v2(v)?),
+	})
+}
+
+pub fn convert_commit_request_v1_to_v2(x: v1::CommitRequest) -> Result<v2::CommitRequest> {
+	Ok(v2::CommitRequest {
+		read_version: x.read_version,
+		conflict_ranges: x
+			.conflict_ranges
+			.into_iter()
+			.map(|v| convert_conflict_range_v1_to_v2(v))
+			.collect::<Result<Vec<_>>>()?,
+		operations: x
+			.operations
+			.into_iter()
+			.map(|v| convert_operation_v1_to_v2(v))
+			.collect::<Result<Vec<_>>>()?,
+		client_node_id: x.client_node_id,
+		client_seq: x.client_seq,
+	})
+}
+
+pub fn convert_commit_committed_v1_to_v2(x: v1::CommitCommitted) -> Result<v2::CommitCommitted> {
+	Ok(v2::CommitCommitted {
+		commit_version: x.commit_version,
+	})
+}
+
+pub fn convert_commit_reply_v1_to_v2(x: v1::CommitReply) -> Result<v2::CommitReply> {
+	Ok(match x {
+		v1::CommitReply::CommitCommitted(v) => {
+			v2::CommitReply::CommitCommitted(convert_commit_committed_v1_to_v2(v)?)
+		}
+		v1::CommitReply::CommitConflict => v2::CommitReply::CommitConflict,
+	})
+}
+
+pub fn convert_watermark_v1_to_v2(x: v1::Watermark) -> Result<v2::Watermark> {
+	Ok(v2::Watermark {
+		durable_version: x.durable_version,
+	})
+}
diff --git a/engine/sdks/rust/universaldb-commit/src/versioned/v2_to_v1.rs b/engine/sdks/rust/universaldb-commit/src/versioned/v2_to_v1.rs
new file mode 100644
index 0000000000..bc1280019f
--- /dev/null
+++ b/engine/sdks/rust/universaldb-commit/src/versioned/v2_to_v1.rs
@@ -0,0 +1,116 @@
+#![allow(dead_code, unused_variables)]
+
+use anyhow::Result;
+
+use crate::generated::{v1, v2};
+
+pub fn convert_conflict_range_type_v2_to_v1(
+	x: v2::ConflictRangeType,
+) -> Result<v1::ConflictRangeType> {
+	Ok(match x {
+		v2::ConflictRangeType::Read => v1::ConflictRangeType::Read,
+		v2::ConflictRangeType::Write => v1::ConflictRangeType::Write,
+	})
+}
+
+pub fn convert_conflict_range_v2_to_v1(x: v2::ConflictRange) -> Result<v1::ConflictRange> {
+	Ok(v1::ConflictRange {
+		begin: x.begin,
+		end: x.end,
+		kind: convert_conflict_range_type_v2_to_v1(x.kind)?,
+	})
+}
+
+pub fn convert_mutation_type_v2_to_v1(x: v2::MutationType) -> Result<v1::MutationType> {
+	Ok(match x {
+		v2::MutationType::Add => v1::MutationType::Add,
+		v2::MutationType::And => v1::MutationType::And,
+		v2::MutationType::BitAnd => v1::MutationType::BitAnd,
+		v2::MutationType::Or => v1::MutationType::Or,
+		v2::MutationType::BitOr => v1::MutationType::BitOr,
+		v2::MutationType::Xor => v1::MutationType::Xor,
+		v2::MutationType::BitXor => v1::MutationType::BitXor,
+		v2::MutationType::AppendIfFits => v1::MutationType::AppendIfFits,
+		v2::MutationType::Max => v1::MutationType::Max,
+		v2::MutationType::Min => v1::MutationType::Min,
+		v2::MutationType::SetVersionstampedKey => v1::MutationType::SetVersionstampedKey,
+		v2::MutationType::SetVersionstampedValue => v1::MutationType::SetVersionstampedValue,
+		v2::MutationType::ByteMin => v1::MutationType::ByteMin,
+		v2::MutationType::ByteMax => v1::MutationType::ByteMax,
+		v2::MutationType::CompareAndClear => v1::MutationType::CompareAndClear,
+	})
+}
+
+pub fn convert_set_value_v2_to_v1(x: v2::SetValue) -> Result<v1::SetValue> {
+	Ok(v1::SetValue {
+		key: x.key,
+		value: x.value,
+	})
+}
+
+pub fn convert_clear_v2_to_v1(x: v2::Clear) -> Result<v1::Clear> {
+	Ok(v1::Clear { key: x.key })
+}
+
+pub fn convert_clear_range_v2_to_v1(x: v2::ClearRange) -> Result<v1::ClearRange> {
+	Ok(v1::ClearRange {
+		begin: x.begin,
+		end: x.end,
+	})
+}
+
+pub fn convert_atomic_op_v2_to_v1(x: v2::AtomicOp) -> Result<v1::AtomicOp> {
+	Ok(v1::AtomicOp {
+		key: x.key,
+		param: x.param,
+		op_type: convert_mutation_type_v2_to_v1(x.op_type)?,
+	})
+}
+
+pub fn convert_operation_v2_to_v1(x: v2::Operation) -> Result<v1::Operation> {
+	Ok(match x {
+		v2::Operation::SetValue(v) => v1::Operation::SetValue(convert_set_value_v2_to_v1(v)?),
+		v2::Operation::Clear(v) => v1::Operation::Clear(convert_clear_v2_to_v1(v)?),
+		v2::Operation::ClearRange(v) => v1::Operation::ClearRange(convert_clear_range_v2_to_v1(v)?),
+		v2::Operation::AtomicOp(v) => v1::Operation::AtomicOp(convert_atomic_op_v2_to_v1(v)?),
+	})
+}
+
+pub fn convert_commit_request_v2_to_v1(x: v2::CommitRequest) -> Result<v1::CommitRequest> {
+	Ok(v1::CommitRequest {
+		read_version: x.read_version,
+		conflict_ranges: x
+			.conflict_ranges
+			.into_iter()
+			.map(|v| convert_conflict_range_v2_to_v1(v))
+			.collect::<Result<Vec<_>>>()?,
+		operations: x
+			.operations
+			.into_iter()
+			.map(|v| convert_operation_v2_to_v1(v))
+			.collect::<Result<Vec<_>>>()?,
+		client_node_id: x.client_node_id,
+		client_seq: x.client_seq,
+	})
+}
+
+pub fn convert_commit_committed_v2_to_v1(x: v2::CommitCommitted) -> Result<v1::CommitCommitted> {
+	Ok(v1::CommitCommitted {
+		commit_version: x.commit_version,
+	})
+}
+
+pub fn convert_commit_reply_v2_to_v1(x: v2::CommitReply) -> Result<v1::CommitReply> {
+	Ok(match x {
+		v2::CommitReply::CommitCommitted(v) => {
+			v1::CommitReply::CommitCommitted(convert_commit_committed_v2_to_v1(v)?)
+		}
+		v2::CommitReply::CommitConflict => v1::CommitReply::CommitConflict,
+	})
+}
+
+pub fn convert_watermark_v2_to_v1(x: v2::Watermark) -> Result<v1::Watermark> {
+	Ok(v1::Watermark {
+		durable_version: x.durable_version,
+	})
+}
diff --git a/engine/sdks/schemas/universaldb-commit/v2.bare b/engine/sdks/schemas/universaldb-commit/v2.bare
new file mode 100644
index 0000000000..3ac2155e29
--- /dev/null
+++ b/engine/sdks/schemas/universaldb-commit/v2.bare
@@ -0,0 +1,114 @@
+# Commit wire format for the multi-node Postgres leader-resolver UDB driver.
+#
+# In multi-node mode a follower encodes a CommitRequest and sends it to the
+# elected leader over NATS request/reply; the leader decodes it to resolve and
+# apply. Rust-only (never leaves the engine), but versioned so rolling deploys
+# can skew follower vs leader code.
+
+type ConflictRangeType enum {
+	READ
+	WRITE
+}
+
+type ConflictRange struct {
+	begin: data
+	end: data
+	kind: ConflictRangeType
+}
+
+# Order MUST match universaldb::options::MutationType declaration order so the
+# enum tag round-trips. 15 variants today; never reorder, only append.
+type MutationType enum {
+	ADD
+	AND
+	BIT_AND
+	OR
+	BIT_OR
+	XOR
+	BIT_XOR
+	APPEND_IF_FITS
+	MAX
+	MIN
+	SET_VERSIONSTAMPED_KEY
+	SET_VERSIONSTAMPED_VALUE
+	BYTE_MIN
+	BYTE_MAX
+	COMPARE_AND_CLEAR
+}
+
+type SetValue struct {
+	key: data
+	value: data
+}
+
+type Clear struct {
+	key: data
+}
+
+type ClearRange struct {
+	begin: data
+	end: data
+}
+
+type AtomicOp struct {
+	key: data
+	param: data
+	opType: MutationType
+}
+
+type Operation union {
+	SetValue |
+	Clear |
+	ClearRange |
+	AtomicOp
+}
+
+type CommitRequest struct {
+	readVersion: u64
+	conflictRanges: list<ConflictRange>
+	operations: list<Operation>
+	# Failover dedup key. `clientNodeId` is the submitting follower's node id and
+	# `clientSeq` a per-process monotonic counter, unique per logical commit. The
+	# leader records committed (clientNodeId, clientSeq) so a follower that resends
+	# the same commit after an indeterminate leader failure is applied exactly once.
+	clientNodeId: data
+	clientSeq: u64
+}
+
+# One piece of an encoded CommitRequest that is larger than the NATS server's
+# max_payload. A follower splits the embedded-version CommitRequest bytes into
+# `count` consecutive pieces and sends them in order on the leader's chunk
+# subject; the last piece carries the NATS reply inbox, and the leader decodes
+# the reassembled bytes as a CommitRequest once it has every piece.
+type CommitRequestChunk struct {
+	clientNodeId: data
+	clientSeq: u64
+	# Incremented on every resend of the same commit, so the leader discards the
+	# pieces of an earlier attempt instead of mixing them into this one.
+	attempt: u32
+	index: u32
+	count: u32
+	data: data
+}
+
+# Reply the leader sends back to the follower over the NATS request/reply inbox
+# after resolving a commit. `CommitCommitted` carries the assigned commit
+# version; `CommitConflict` (read-write conflict or cold-window reject) carries
+# nothing and maps to the retryable NotCommitted on the follower.
+type CommitCommitted struct {
+	commitVersion: i64
+}
+
+type CommitConflict void
+
+type CommitReply union {
+	CommitCommitted |
+	CommitConflict
+}
+
+# Durable-version watermark the leader broadcasts on every applied batch so
+# followers advance their read-snapshot floor. Best-effort; a lease-row poll is
+# the backstop.
+type Watermark struct {
+	durableVersion: i64
+}