From 8a1bfcbcdfcc785ca2b2f402d19e695792568e58 Mon Sep 17 00:00:00 2001
From: Randy <857271370@qq.com>
Date: Sun, 8 Nov 2026 10:29:13 +0800
Subject: [PATCH 1/6] =?UTF-8?q?=E2=9C=A8=20feat:=20fold=20MongoDB=20puts,?=
 =?UTF-8?q?=20deletes=20and=20Lua=20programs=20of=20one=20record?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 docs/batching.md                              |   6 +
 docs/lua.md                                   |   4 +-
 docs/operations.md                            |   4 +-
 internal/backend/mongodb/adapter.go           |   7 +
 internal/backend/mongodb/batch.go             |  28 ++-
 internal/backend/mongodb/fold.go              | 214 +++++++++++++++++
 .../backend/mongodb/fold_integration_test.go  | 220 ++++++++++++++++++
 internal/backend/mongodb/program.go           |  50 +++-
 8 files changed, 518 insertions(+), 15 deletions(-)
 create mode 100644 internal/backend/mongodb/fold.go
 create mode 100644 internal/backend/mongodb/fold_integration_test.go

diff --git a/docs/batching.md b/docs/batching.md
index 7eac33e3..f5101a34 100644
--- a/docs/batching.md
+++ b/docs/batching.md
@@ -200,6 +200,12 @@ first mutation whose caller remains is a put or delete. Query-visible mutations
 wait for one refresh, or each for its own when they run one at a time. One whose
 precondition fails or whose Lua rejects leaves the state to the next.
 
+MongoDB folds puts, deletes and Lua programs. A record's puts and deletes become
+the last of them, which leaves the record as the whole run would. Its Lua
+programs evaluate in order on the transaction's one read of the record, each on
+the document the programs before it left, and the transaction writes the result
+once. Creates, replaces and backend expressions do not fold.
+
 Grouping is not an application transaction:
 a MongoDB Lua transaction may include records from several RPCs, and its physical
 membership varies with arrivals and available capacity.
diff --git a/docs/lua.md b/docs/lua.md
index 32bebddc..58097b05 100644
--- a/docs/lua.md
+++ b/docs/lua.md
@@ -140,6 +140,8 @@ final process boundary.
 Confirmed MongoDB transaction and Search compare-and-swap conflicts can trigger
 a bounded fresh read and reevaluation. Search evaluates the queued mutations of
 one record in queue order on one read and writes once, so each Lua callback sees
-the result of the mutations before it, and a conflict reevaluates them all. Ambiguous write or commit outcomes never
+the result of the mutations before it, and a conflict reevaluates them all.
+MongoDB does the same for the Lua programs of one record in its transaction.
+Ambiguous write or commit outcomes never
 replay the business callback. A callback error or invalid document does not
 write.
diff --git a/docs/operations.md b/docs/operations.md
index 7d104238..2bed6815 100644
--- a/docs/operations.md
+++ b/docs/operations.md
@@ -254,8 +254,8 @@ they form larger batches, and more Weir replicas add no capacity. The ratio is 0
 when `min_in_flight` equals `max_in_flight`, and it keeps its last value while a
 Store is idle. Free slots mean records wait for earlier
 mutations of the same resource: one stream's mutations of a resource run one at
-a time unless the adapter folds them, as Search does for all but backend
-expressions, so a client that keeps rewriting the same records queues behind
+a time unless the adapter [folds](batching.md) them, so a client that keeps
+rewriting the same records with mutations that do not fold queues behind
 itself.
 
 ### WeirWatchdogExpirations
diff --git a/internal/backend/mongodb/adapter.go b/internal/backend/mongodb/adapter.go
index 199d0359..cae98c32 100644
--- a/internal/backend/mongodb/adapter.go
+++ b/internal/backend/mongodb/adapter.go
@@ -214,9 +214,14 @@ func (a *Adapter) PrepareRecord(ctx context.Context, record *execution.Record) (
 		return work, nil
 	}
 	work.BatchKey = "records"
+	// Puts and deletes of one record fold into the last, which leaves it as
+	// they all would; Lua programs of one record fold into one transaction
+	// write. Creates, replaces and expressions depend on a state no write of
+	// this batch reads.
 	switch action := mutation.Action.(type) {
 	case *pb.MutateRequest_Put:
 		native.action = "put"
+		work.Folds = true
 		return work, native.setDocument(action.Put)
 	case *pb.MutateRequest_Create:
 		native.action = "create"
@@ -226,12 +231,14 @@ func (a *Adapter) PrepareRecord(ctx context.Context, record *execution.Record) (
 		return work, native.setDocument(action.Replace)
 	case *pb.MutateRequest_Delete:
 		native.action = "delete"
+		work.Folds = true
 		return work, nil
 	case *pb.MutateRequest_AtomicTransform:
 		if program := action.AtomicTransform.GetLua(); program != nil {
 			native.action = "program"
 			native.program = &luaengine.Program{Source: string(program.Source), ObservedAt: time.Now()}
 			work.BatchKey = target.String() + "/lua"
+			work.Folds = true
 			return work, validateProgramInput(ctx, program.Input)
 		}
 		expression := action.AtomicTransform.GetBackendExpression()
diff --git a/internal/backend/mongodb/batch.go b/internal/backend/mongodb/batch.go
index 0754f891..b52618a0 100644
--- a/internal/backend/mongodb/batch.go
+++ b/internal/backend/mongodb/batch.go
@@ -3,6 +3,7 @@ package mongodb
 import (
 	"context"
 	"errors"
+	"slices"
 	"strconv"
 
 	"github.com/batchstream/weir-protocol/api/protocol"
@@ -268,6 +269,7 @@ func rawRecordID(raw bson.RawValue) (any, bool) {
 }
 
 func (a *Adapter) executeWrites(ctx context.Context, plans []*execution.Plan) []*pb.Event {
+	base := ctx
 	results := make([]*pb.Event, len(plans))
 	active := make([]*execution.Plan, 0, len(plans))
 	positions := make([]int, 0, len(plans))
@@ -295,6 +297,8 @@ func (a *Adapter) executeWrites(ctx context.Context, plans []*execution.Plan) []
 	if len(active) == 0 {
 		return results
 	}
+	live, livePositions := slices.Clone(active), slices.Clone(positions)
+	active, positions, writes, stands := foldWrites(active, positions, writes)
 
 	encoded := encodeBulkWrite(writes, namespaces)
 	if len(plans) > 1 && len(encoded) > a.commandBytes() {
@@ -330,10 +334,10 @@ func (a *Adapter) executeWrites(ctx context.Context, plans []*execution.Plan) []
 		// operation. Only this definite initial rejection permits isolation.
 		groups := make([][]*execution.Plan, len(namespaces))
 		groupPositions := make([][]int, len(namespaces))
-		for i, p := range active {
+		for i, p := range live {
 			index := namespaceIndexes[p.Backend.(*plan).target]
 			groups[index] = append(groups[index], p)
-			groupPositions[index] = append(groupPositions[index], positions[i])
+			groupPositions[index] = append(groupPositions[index], livePositions[i])
 		}
 		for i, group := range groups {
 			if len(group) == 0 {
@@ -368,9 +372,23 @@ func (a *Adapter) executeWrites(ctx context.Context, plans []*execution.Plan) []
 		if reply == nil {
 			reply = protocol.Mutation(pb.MutationOutcome_UNKNOWN, protocol.Fail(pb.FailureCode_UNAVAILABLE, "write acknowledgement unavailable or incomplete"))
 		}
-		value := &pb.Event_MutationResult{MutationResult: reply}
-		event := &pb.Event{Value: value}
-		results[positions[i]] = event
+		members := stands[i]
+		if members != nil && reply.Outcome == pb.MutationOutcome_NOT_APPLIED {
+			// Nothing of the chain applied, so each write runs alone.
+			for _, position := range members {
+				single := []*execution.Plan{plans[position]}
+				results[position] = a.executeWrites(base, single)[0]
+			}
+			continue
+		}
+		if members == nil {
+			members = positions[i : i+1]
+		}
+		for _, position := range members {
+			value := &pb.Event_MutationResult{MutationResult: reply}
+			event := &pb.Event{Value: value}
+			results[position] = event
+		}
 	}
 	return results
 }
diff --git a/internal/backend/mongodb/fold.go b/internal/backend/mongodb/fold.go
new file mode 100644
index 00000000..3798bc07
--- /dev/null
+++ b/internal/backend/mongodb/fold.go
@@ -0,0 +1,214 @@
+package mongodb
+
+import (
+	"bytes"
+	"context"
+	"sync/atomic"
+	"time"
+
+	"github.com/batchstream/weir-protocol/api/protocol"
+	pb "github.com/batchstream/weir-protocol/api/weir/v1"
+	"github.com/batchstream/weir/internal/execution"
+	"go.mongodb.org/mongo-driver/v2/bson"
+)
+
+// programFolds holds the chains of Lua programs of one record that a program
+// batch runs as one, each under a plan that stands for the whole chain.
+type programFolds struct {
+	chains map[*execution.Plan][]*execution.Plan
+	// own holds, from a chain's last evaluation, the result of each program
+	// that did not take part in its write, and nil for those that did.
+	own map[*execution.Plan][]*pb.MutationResult
+	// unfold marks chains whose write MongoDB rejected; none of their
+	// programs applied, so they run again one at a time.
+	unfold   map[*execution.Plan]bool
+	releases []func()
+}
+
+func (f *programFolds) chain(stand *execution.Plan) []*execution.Plan {
+	if f == nil {
+		return nil
+	}
+	return f.chains[stand]
+}
+
+func (f *programFolds) release() {
+	for _, release := range f.releases {
+		release()
+	}
+}
+
+// foldPrograms links the programs of one record into chains and returns the
+// plans a program batch runs: each single program, and for each chain a plan
+// whose caller remains while any of its programs' callers does.
+func foldPrograms(plans []*execution.Plan) ([]*execution.Plan, *programFolds) {
+	members := make(map[string][]*execution.Plan, len(plans))
+	order := make([]string, 0, len(plans))
+	for _, work := range plans {
+		if _, seen := members[work.Key]; !seen {
+			order = append(order, work.Key)
+		}
+		members[work.Key] = append(members[work.Key], work)
+	}
+	if len(order) == len(plans) {
+		return plans, nil
+	}
+	folds := &programFolds{
+		chains: make(map[*execution.Plan][]*execution.Plan),
+		own:    make(map[*execution.Plan][]*pb.MutationResult),
+		unfold: make(map[*execution.Plan]bool),
+	}
+	stands := make([]*execution.Plan, 0, len(order))
+	for _, key := range order {
+		chain := members[key]
+		if len(chain) == 1 || !chain[0].Folds {
+			stands = append(stands, chain...)
+			continue
+		}
+		stand := *chain[0]
+		ctx, release := chainContext(chain)
+		stand.Context = ctx
+		folds.releases = append(folds.releases, release)
+		folds.chains[&stand] = chain
+		stands = append(stands, &stand)
+	}
+	return stands, folds
+}
+
+// chainContext ends once every member's context has ended, and no later than
+// the latest member deadline when every member has one.
+func chainContext(chain []*execution.Plan) (context.Context, func()) {
+	var latest time.Time
+	bounded := true
+	for _, work := range chain {
+		deadline, ok := work.Context.Deadline()
+		bounded = bounded && ok
+		if deadline.After(latest) {
+			latest = deadline
+		}
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	if bounded {
+		ctx, cancel = context.WithDeadline(context.Background(), latest)
+	}
+	var live atomic.Int64
+	live.Store(int64(len(chain)))
+	stops := make([]func() bool, 0, len(chain))
+	for _, work := range chain {
+		stop := context.AfterFunc(work.Context, func() {
+			if live.Add(-1) == 0 {
+				cancel()
+			}
+		})
+		stops = append(stops, stop)
+	}
+	release := func() {
+		for _, stop := range stops {
+			stop()
+		}
+		cancel()
+	}
+	return ctx, release
+}
+
+// transformChain evaluates a chain's programs in queue order, each on the
+// document the programs before it left, and returns one write from the read
+// document to the last. A program that does not change the document, or that
+// fails, leaves it to the next.
+func (a *Adapter) transformChain(batch context.Context, folds *programFolds, stand *execution.Plan, raw bson.Raw) (*execution.Plan, *pb.MutationResult) {
+	chain := folds.chains[stand]
+	own := make([]*pb.MutationResult, len(chain))
+	folds.own[stand] = own
+	state := raw
+	for i, work := range chain {
+		write, result := a.transformProgram(batch, work, state)
+		switch {
+		case result == nil && write.Backend.(*plan).action == "delete":
+			state = nil
+		case result == nil:
+			state = write.Backend.(*plan).document
+		case result.Outcome != pb.MutationOutcome_APPLIED || result.Failure != nil:
+			own[i] = result
+		}
+	}
+	native := stand.Backend.(*plan)
+	prepared := &plan{target: native.target, id: native.id}
+	switch {
+	case raw == nil && state == nil, raw != nil && state != nil && bytes.Equal(raw, state):
+		unchanged := protocol.Mutation(pb.MutationOutcome_APPLIED, nil)
+		return nil, unchanged
+	case state == nil:
+		prepared.action = "delete"
+	case raw == nil:
+		prepared.action, prepared.document = "create", state
+	default:
+		prepared.action, prepared.document = "replace", state
+	}
+	write := &execution.Plan{Command: stand.Command, Backend: prepared}
+	return write, nil
+}
+
+// unfoldPrograms runs each program of a rejected chain alone, in queue order.
+func (a *Adapter) unfoldPrograms(ctx context.Context, chain []*execution.Plan) []*pb.MutationResult {
+	results := make([]*pb.MutationResult, len(chain))
+	for i, work := range chain {
+		single := []*execution.Plan{work}
+		batch := &programBatch{plans: single, results: make([]*pb.MutationResult, 1)}
+		a.runPrograms(ctx, batch)
+		results[i] = batch.results[0]
+	}
+	return results
+}
+
+// chainResult is a program's result from its chain's result: a caller that
+// left before the chain ran keeps its own cause.
+func chainResult(work *execution.Plan, result *pb.MutationResult) *pb.MutationResult {
+	code := result.GetFailure().GetCode()
+	unstarted := result.GetOutcome() == pb.MutationOutcome_NOT_STARTED || result.GetOutcome() == pb.MutationOutcome_NOT_APPLIED
+	if unstarted && work.Context.Err() != nil && (code == pb.FailureCode_CANCELLED || code == pb.FailureCode_DEADLINE_EXCEEDED) {
+		own := protocol.Mutation(result.GetOutcome(), protocol.ContextFailure(work.Context))
+		return own
+	}
+	return result
+}
+
+// foldWrites keeps, of each chain of puts and deletes of one record, the last
+// live one, which leaves the record as the whole chain would since each
+// replaces or removes it whatever it held. It returns the kept writes and,
+// for each kept write that stands for others, all their positions.
+func foldWrites(active []*execution.Plan, positions []int, writes []bulkWrite) ([]*execution.Plan, []int, []bulkWrite, map[int][]int) {
+	indexes := make(map[string][]int, len(active))
+	for i, work := range active {
+		if work.Folds {
+			indexes[work.Key] = append(indexes[work.Key], i)
+		}
+	}
+	chains := make(map[string][]int)
+	for key, chain := range indexes {
+		if len(chain) < 2 {
+			continue
+		}
+		members := make([]int, len(chain))
+		for j, i := range chain {
+			members[j] = positions[i]
+		}
+		chains[key] = members
+	}
+	if len(chains) == 0 {
+		return active, positions, writes, nil
+	}
+	kept := 0
+	stands := make(map[int][]int, len(chains))
+	for i, work := range active {
+		if members := chains[work.Key]; work.Folds && members != nil {
+			chain := indexes[work.Key]
+			if chain[len(chain)-1] != i {
+				continue
+			}
+			stands[kept] = members
+		}
+		active[kept], positions[kept], writes[kept] = work, positions[i], writes[i]
+		kept++
+	}
+	return active[:kept], positions[:kept], writes[:kept], stands
+}
diff --git a/internal/backend/mongodb/fold_integration_test.go b/internal/backend/mongodb/fold_integration_test.go
new file mode 100644
index 00000000..dcdb98f4
--- /dev/null
+++ b/internal/backend/mongodb/fold_integration_test.go
@@ -0,0 +1,220 @@
+//go:build integration
+
+package mongodb
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+
+	pb "github.com/batchstream/weir-protocol/api/weir/v1"
+	"github.com/batchstream/weir/internal/execution"
+	"github.com/batchstream/weir/internal/testutil/testmongo"
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/event"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+	"go.mongodb.org/mongo-driver/v2/mongo/options"
+)
+
+const rejectProgram = `return function(current, incoming) return weir.reject("business rule") end`
+
+func setProgram(n string) string {
+	program := `return function(current, incoming) return {n = ` + n + `} end`
+	return program
+}
+
+// foldCommands counts the commands a folded batch sends.
+type foldCommands struct {
+	finds, writes, commits, ops atomic.Int32
+}
+
+func (c *foldCommands) monitor() *event.CommandMonitor {
+	monitor := &event.CommandMonitor{Started: func(_ context.Context, e *event.CommandStartedEvent) {
+		switch e.CommandName {
+		case "find":
+			c.finds.Add(1)
+		case "bulkWrite":
+			c.writes.Add(1)
+			ops, _ := e.Command.Lookup("ops").Array().Values()
+			c.ops.Add(int32(len(ops)))
+		case "commitTransaction":
+			c.commits.Add(1)
+		}
+	}}
+	return monitor
+}
+
+// recordWrites prepares mutations of records of one collection.
+type recordWrites struct {
+	t       *testing.T
+	adapter *Adapter
+	prefix  string
+}
+
+func (r recordWrites) program(id, source string) *execution.Plan {
+	opts := batchOperationOptions{resource: r.prefix + id, index: 1, program: source}
+	return prepareBatchProgram(r.t, r.adapter, opts)
+}
+
+// write prepares a put or delete of the record id; a put stores n.
+func (r recordWrites) write(id, action string, n int64) *execution.Plan {
+	r.t.Helper()
+	var document bson.D
+	if action != "delete" {
+		document = bson.D{{Key: "_id", Value: id}, {Key: "n", Value: n}}
+	}
+	opts := batchOperationOptions{resource: r.prefix + id, action: action, document: document}
+	work, failure := prepareTestRecord(r.adapter, batchOperation(r.t, opts))
+	if failure != nil {
+		r.t.Fatal(failure)
+	}
+	work.Context = r.t.Context()
+	return work
+}
+
+func checkOutcomes(t *testing.T, results []*pb.Event, want []pb.MutationOutcome) {
+	t.Helper()
+	for i, result := range results {
+		if got := result.GetMutationResult().GetOutcome(); got != want[i] {
+			t.Fatalf("result %d: %v, want %v", i, result.GetMutationResult(), want[i])
+		}
+	}
+}
+
+func storedN(t *testing.T, collection *mongo.Collection, id string) (int64, bool) {
+	t.Helper()
+	var document struct{ N int64 }
+	filter := bson.D{{Key: "_id", Value: id}}
+	err := collection.FindOne(t.Context(), filter).Decode(&document)
+	if err == mongo.ErrNoDocuments {
+		return 0, false
+	}
+	if err != nil {
+		t.Fatal(err)
+	}
+	return document.N, true
+}
+
+func TestMongoLuaProgramsOfOneRecordFoldIntoOneTransactionWrite(t *testing.T) {
+	fixture := testmongo.Open(t)
+	collection := fixture.Admin.Database(fixture.DB).Collection("records")
+	for _, id := range []string{"a", "gone"} {
+		document := bson.D{{Key: "_id", Value: id}, {Key: "n", Value: int64(1)}}
+		if _, err := collection.InsertOne(t.Context(), document); err != nil {
+			t.Fatal(err)
+		}
+	}
+	commands := &foldCommands{}
+	adapterOpts := adapterTestOptions{fixture: fixture, monitor: commands.monitor()}
+	adapter := testAdapter(t, adapterOpts)
+	records := recordWrites{t: t, adapter: adapter, prefix: fixture.DB + "/records/s:"}
+	plans := []*execution.Plan{
+		records.program("a", incrementProgram),
+		records.program("new", incrementProgram),
+		records.program("a", incrementProgram),
+		records.program("gone", incrementProgram),
+		records.program("a", rejectProgram),
+		records.program("new", incrementProgram),
+		records.program("a", incrementProgram),
+		records.program("gone", `return function(current, incoming) return weir.delete() end`),
+	}
+	results := adapter.executePrograms(t.Context(), plans)
+	applied, rejected := pb.MutationOutcome_APPLIED, pb.MutationOutcome_NOT_APPLIED
+	want := []pb.MutationOutcome{applied, applied, applied, applied, rejected, applied, applied, applied}
+	checkOutcomes(t, results, want)
+	if commands.finds.Load() != 1 || commands.writes.Load() != 1 || commands.ops.Load() != 3 || commands.commits.Load() != 1 {
+		t.Fatal("finds", commands.finds.Load(), "writes", commands.writes.Load(), "ops", commands.ops.Load(), "commits", commands.commits.Load())
+	}
+	if n, found := storedN(t, collection, "a"); !found || n != 4 {
+		t.Fatal("folded increments", n, found)
+	}
+	if n, found := storedN(t, collection, "new"); !found || n != 2 {
+		t.Fatal("folded create", n, found)
+	}
+	if _, found := storedN(t, collection, "gone"); found {
+		t.Fatal("folded delete left the record")
+	}
+}
+
+func TestMongoRejectedFoldedProgramsRunOneAtATime(t *testing.T) {
+	fixture := testmongo.Open(t)
+	database := fixture.Admin.Database(fixture.DB)
+	validator := bson.D{{Key: "n", Value: bson.D{{Key: "$lte", Value: 10}}}}
+	create := options.CreateCollection().SetValidator(validator)
+	if err := database.CreateCollection(t.Context(), "validated", create); err != nil {
+		t.Fatal(err)
+	}
+	adapterOpts := adapterTestOptions{fixture: fixture}
+	adapter := testAdapter(t, adapterOpts)
+	records := recordWrites{t: t, adapter: adapter, prefix: fixture.DB + "/validated/s:"}
+	plans := []*execution.Plan{
+		records.program("v", setProgram("5")),
+		records.program("v", setProgram("50")),
+	}
+	results := adapter.executePrograms(t.Context(), plans)
+	want := []pb.MutationOutcome{pb.MutationOutcome_APPLIED, pb.MutationOutcome_NOT_APPLIED}
+	checkOutcomes(t, results, want)
+	if code := results[1].GetMutationResult().GetFailure().GetCode(); code != pb.FailureCode_PRECONDITION_FAILED {
+		t.Fatal("rejected program failure", code)
+	}
+	if n, found := storedN(t, database.Collection("validated"), "v"); !found || n != 5 {
+		t.Fatal("the program before the rejected one did not apply alone", n, found)
+	}
+}
+
+func TestMongoPutsAndDeletesOfOneRecordFoldIntoTheLast(t *testing.T) {
+	fixture := testmongo.Open(t)
+	collection := fixture.Admin.Database(fixture.DB).Collection("records")
+	document := bson.D{{Key: "_id", Value: "y"}, {Key: "n", Value: int64(1)}}
+	if _, err := collection.InsertOne(t.Context(), document); err != nil {
+		t.Fatal(err)
+	}
+	commands := &foldCommands{}
+	adapterOpts := adapterTestOptions{fixture: fixture, monitor: commands.monitor()}
+	adapter := testAdapter(t, adapterOpts)
+	records := recordWrites{t: t, adapter: adapter, prefix: fixture.DB + "/records/s:"}
+	plans := []*execution.Plan{
+		records.write("x", "put", 1),
+		records.write("y", "delete", 0),
+		records.write("x", "put", 2),
+		records.write("z", "delete", 0),
+		records.write("y", "put", 3),
+		records.write("z", "delete", 0),
+	}
+	results := adapter.ExecuteRecords(t.Context(), plans)
+	applied := pb.MutationOutcome_APPLIED
+	want := []pb.MutationOutcome{applied, applied, applied, applied, applied, applied}
+	checkOutcomes(t, results, want)
+	if commands.writes.Load() != 1 || commands.ops.Load() != 3 {
+		t.Fatal("writes", commands.writes.Load(), "ops", commands.ops.Load())
+	}
+	if n, found := storedN(t, collection, "x"); !found || n != 2 {
+		t.Fatal("folded puts", n, found)
+	}
+	if n, found := storedN(t, collection, "y"); !found || n != 3 {
+		t.Fatal("folded delete and put", n, found)
+	}
+}
+
+func TestMongoRejectedFoldedPutsRunOneAtATime(t *testing.T) {
+	fixture := testmongo.Open(t)
+	database := fixture.Admin.Database(fixture.DB)
+	validator := bson.D{{Key: "n", Value: bson.D{{Key: "$lte", Value: 10}}}}
+	create := options.CreateCollection().SetValidator(validator)
+	if err := database.CreateCollection(t.Context(), "validated", create); err != nil {
+		t.Fatal(err)
+	}
+	adapterOpts := adapterTestOptions{fixture: fixture}
+	adapter := testAdapter(t, adapterOpts)
+	records := recordWrites{t: t, adapter: adapter, prefix: fixture.DB + "/validated/s:"}
+	plans := []*execution.Plan{
+		records.write("v", "put", 1),
+		records.write("v", "put", 50),
+	}
+	results := adapter.ExecuteRecords(t.Context(), plans)
+	want := []pb.MutationOutcome{pb.MutationOutcome_APPLIED, pb.MutationOutcome_NOT_APPLIED}
+	checkOutcomes(t, results, want)
+	if n, found := storedN(t, database.Collection("validated"), "v"); !found || n != 1 {
+		t.Fatal("the put before the rejected one did not apply alone", n, found)
+	}
+}
diff --git a/internal/backend/mongodb/program.go b/internal/backend/mongodb/program.go
index 298d7fab..d010b84e 100644
--- a/internal/backend/mongodb/program.go
+++ b/internal/backend/mongodb/program.go
@@ -28,17 +28,44 @@ type programBatch struct {
 	// attempted records that a transaction exchange has run, so members that
 	// end later report NOT_APPLIED rather than NOT_STARTED.
 	attempted bool
+	folds     *programFolds
 }
 
-// Programs of a batch name one collection and distinct records, so they share a
-// snapshot transaction. Caller contexts govern evaluation; the batch context
+// Programs of a batch name one collection and share a snapshot transaction.
+// Programs of one record run as one chain: one read, evaluated in queue
+// order, and one write. Caller contexts govern evaluation; the batch context
 // governs I/O.
 func (a *Adapter) executePrograms(ctx context.Context, plans []*execution.Plan) []*pb.Event {
-	batch := &programBatch{plans: plans, results: make([]*pb.MutationResult, len(plans))}
+	stands, folds := foldPrograms(plans)
+	if folds != nil {
+		defer folds.release()
+	}
+	batch := &programBatch{plans: stands, results: make([]*pb.MutationResult, len(stands)), folds: folds}
 	a.runPrograms(ctx, batch)
+	byPlan := make(map[*execution.Plan]*pb.MutationResult, len(plans))
+	for i, stand := range stands {
+		chain := folds.chain(stand)
+		switch {
+		case chain == nil:
+			byPlan[stand] = batch.results[i]
+		case folds.unfold[stand]:
+			for j, result := range a.unfoldPrograms(ctx, chain) {
+				byPlan[chain[j]] = result
+			}
+		default:
+			own := folds.own[stand]
+			for j, work := range chain {
+				result := chainResult(work, batch.results[i])
+				if own != nil && own[j] != nil {
+					result = own[j]
+				}
+				byPlan[work] = result
+			}
+		}
+	}
 	results := make([]*pb.Event, len(plans))
-	for i, result := range batch.results {
-		value := &pb.Event_MutationResult{MutationResult: result}
+	for i, work := range plans {
+		value := &pb.Event_MutationResult{MutationResult: byPlan[work]}
 		event := &pb.Event{Value: value}
 		results[i] = event
 	}
@@ -240,6 +267,9 @@ func (a *Adapter) runProgramAttempt(ctx context.Context, state *programAttempt)
 			for i, reply := range replies {
 				if reply != nil && reply.Outcome == pb.MutationOutcome_NOT_APPLIED {
 					batch.results[writePositions[i]] = reply
+					if stand := batch.plans[writePositions[i]]; batch.folds.chain(stand) != nil {
+						batch.folds.unfold[stand] = true
+					}
 				}
 			}
 			if len(batch.active()) == len(active) {
@@ -324,7 +354,7 @@ func (a *Adapter) splitPrograms(ctx context.Context, batch *programBatch, positi
 		for i, position := range part {
 			plans[i] = batch.plans[position]
 		}
-		child := &programBatch{plans: plans, results: make([]*pb.MutationResult, len(plans)), attempted: batch.attempted}
+		child := &programBatch{plans: plans, results: make([]*pb.MutationResult, len(plans)), attempted: batch.attempted, folds: batch.folds}
 		a.runPrograms(ctx, child)
 		for i, position := range part {
 			batch.results[position] = child.results[i]
@@ -412,7 +442,13 @@ func (a *Adapter) transformPrograms(ctx context.Context, batch *programBatch, po
 	writePositions := make([]int, 0, len(positions))
 	for i, position := range positions {
 		work := batch.plans[position]
-		write, result := a.transformProgram(ctx, work, documents[i])
+		var write *execution.Plan
+		var result *pb.MutationResult
+		if batch.folds.chain(work) != nil {
+			write, result = a.transformChain(ctx, batch.folds, work, documents[i])
+		} else {
+			write, result = a.transformProgram(ctx, work, documents[i])
+		}
 		if result != nil {
 			batch.results[position] = result
 		} else {

From dc6060b447932d763612bf11d1802c005aa1d0be Mon Sep 17 00:00:00 2001
From: Randy <857271370@qq.com>
Date: Sun, 8 Nov 2026 10:42:33 +0800
Subject: [PATCH 2/6] =?UTF-8?q?=F0=9F=90=9B=20fix:=20settle=20folded=20Mon?=
 =?UTF-8?q?goDB=20chains=20whose=20callers=20leave=20or=20whose=20write=20?=
 =?UTF-8?q?fails?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 internal/backend/mongodb/batch.go     |   5 +-
 internal/backend/mongodb/fold.go      |  43 ++++++--
 internal/backend/mongodb/fold_test.go | 139 ++++++++++++++++++++++++++
 internal/backend/mongodb/program.go   |  67 +++++++++----
 4 files changed, 227 insertions(+), 27 deletions(-)
 create mode 100644 internal/backend/mongodb/fold_test.go

diff --git a/internal/backend/mongodb/batch.go b/internal/backend/mongodb/batch.go
index b52618a0..19be008c 100644
--- a/internal/backend/mongodb/batch.go
+++ b/internal/backend/mongodb/batch.go
@@ -373,8 +373,9 @@ func (a *Adapter) executeWrites(ctx context.Context, plans []*execution.Plan) []
 			reply = protocol.Mutation(pb.MutationOutcome_UNKNOWN, protocol.Fail(pb.FailureCode_UNAVAILABLE, "write acknowledgement unavailable or incomplete"))
 		}
 		members := stands[i]
-		if members != nil && reply.Outcome == pb.MutationOutcome_NOT_APPLIED {
-			// Nothing of the chain applied, so each write runs alone.
+		if members != nil && reply.Outcome == pb.MutationOutcome_NOT_APPLIED && reply.GetFailure().GetCode() == pb.FailureCode_PRECONDITION_FAILED {
+			// MongoDB rejected the document the chain left, and nothing of
+			// the chain applied, so each write runs alone.
 			for _, position := range members {
 				single := []*execution.Plan{plans[position]}
 				results[position] = a.executeWrites(base, single)[0]
diff --git a/internal/backend/mongodb/fold.go b/internal/backend/mongodb/fold.go
index 3798bc07..0a4f2c5c 100644
--- a/internal/backend/mongodb/fold.go
+++ b/internal/backend/mongodb/fold.go
@@ -18,8 +18,11 @@ type programFolds struct {
 	chains map[*execution.Plan][]*execution.Plan
 	// own holds, from a chain's last evaluation, the result of each program
 	// that did not take part in its write, and nil for those that did.
-	own map[*execution.Plan][]*pb.MutationResult
-	// unfold marks chains whose write MongoDB rejected; none of their
+	// dependent lists the programs rejected on a document that only that
+	// write makes real.
+	own       map[*execution.Plan][]*pb.MutationResult
+	dependent map[*execution.Plan][]int
+	// unfold marks chains whose document MongoDB rejected; none of their
 	// programs applied, so they run again one at a time.
 	unfold   map[*execution.Plan]bool
 	releases []func()
@@ -32,6 +35,21 @@ func (f *programFolds) chain(stand *execution.Plan) []*execution.Plan {
 	return f.chains[stand]
 }
 
+// departed reports whether a program whose effect a chain's write carries
+// has lost its caller since the chain was evaluated.
+func (f *programFolds) departed(ctx context.Context, stand *execution.Plan) bool {
+	if f == nil {
+		return false
+	}
+	own := f.own[stand]
+	for i, work := range f.chains[stand] {
+		if own[i] == nil && work.Ended(ctx) != nil {
+			return true
+		}
+	}
+	return false
+}
+
 func (f *programFolds) release() {
 	for _, release := range f.releases {
 		release()
@@ -54,9 +72,10 @@ func foldPrograms(plans []*execution.Plan) ([]*execution.Plan, *programFolds) {
 		return plans, nil
 	}
 	folds := &programFolds{
-		chains: make(map[*execution.Plan][]*execution.Plan),
-		own:    make(map[*execution.Plan][]*pb.MutationResult),
-		unfold: make(map[*execution.Plan]bool),
+		chains:    make(map[*execution.Plan][]*execution.Plan),
+		own:       make(map[*execution.Plan][]*pb.MutationResult),
+		dependent: make(map[*execution.Plan][]int),
+		unfold:    make(map[*execution.Plan]bool),
 	}
 	stands := make([]*execution.Plan, 0, len(order))
 	for _, key := range order {
@@ -119,6 +138,7 @@ func (a *Adapter) transformChain(batch context.Context, folds *programFolds, sta
 	chain := folds.chains[stand]
 	own := make([]*pb.MutationResult, len(chain))
 	folds.own[stand] = own
+	var dependent []int
 	state := raw
 	for i, work := range chain {
 		write, result := a.transformProgram(batch, work, state)
@@ -129,12 +149,16 @@ func (a *Adapter) transformChain(batch context.Context, folds *programFolds, sta
 			state = write.Backend.(*plan).document
 		case result.Outcome != pb.MutationOutcome_APPLIED || result.Failure != nil:
 			own[i] = result
+			if work.Ended(batch) == nil && !sameDocument(raw, state) {
+				dependent = append(dependent, i)
+			}
 		}
 	}
+	folds.dependent[stand] = dependent
 	native := stand.Backend.(*plan)
 	prepared := &plan{target: native.target, id: native.id}
 	switch {
-	case raw == nil && state == nil, raw != nil && state != nil && bytes.Equal(raw, state):
+	case sameDocument(raw, state):
 		unchanged := protocol.Mutation(pb.MutationOutcome_APPLIED, nil)
 		return nil, unchanged
 	case state == nil:
@@ -148,6 +172,13 @@ func (a *Adapter) transformChain(batch context.Context, folds *programFolds, sta
 	return write, nil
 }
 
+// sameDocument reports whether two documents, nil for a missing one, are
+// the same.
+func sameDocument(a, b bson.Raw) bool {
+	same := a == nil && b == nil || a != nil && b != nil && bytes.Equal(a, b)
+	return same
+}
+
 // unfoldPrograms runs each program of a rejected chain alone, in queue order.
 func (a *Adapter) unfoldPrograms(ctx context.Context, chain []*execution.Plan) []*pb.MutationResult {
 	results := make([]*pb.MutationResult, len(chain))
diff --git a/internal/backend/mongodb/fold_test.go b/internal/backend/mongodb/fold_test.go
new file mode 100644
index 00000000..18cf4ede
--- /dev/null
+++ b/internal/backend/mongodb/fold_test.go
@@ -0,0 +1,139 @@
+package mongodb
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	pb "github.com/batchstream/weir-protocol/api/weir/v1"
+	"github.com/batchstream/weir/internal/execution"
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/event"
+)
+
+const (
+	setOneProgram      = `return function(current, incoming) current.n = 1; return current end`
+	rejectSetProgram   = `return function(current, incoming) if current.n > 0 then return weir.reject("already set") end; current.m = 1; return current end`
+	loopProgramPattern = `return function(current, incoming) local text = "%s"; while true do local upper = string.upper(text) end end`
+)
+
+// mockPrograms prepares programs of records in db/records under the test's
+// context.
+func mockPrograms(t *testing.T, adapter *Adapter, programs [][2]string) []*execution.Plan {
+	t.Helper()
+	plans := make([]*execution.Plan, len(programs))
+	for i, program := range programs {
+		opts := batchOperationOptions{resource: "db/records/s:" + program[0], action: "program", program: program[1]}
+		work, failure := prepareTestRecord(adapter, batchOperation(t, opts))
+		if failure != nil {
+			t.Fatal(failure)
+		}
+		work.Context = t.Context()
+		plans[i] = work
+	}
+	return plans
+}
+
+func TestMongoFoldedPutsRejectedForAnotherReasonDoNotRunAlone(t *testing.T) {
+	item := bson.D{{Key: "ok", Value: 0}, {Key: "idx", Value: 0}, {Key: "code", Value: 13}}
+	cursor := bson.D{{Key: "id", Value: int64(0)}, {Key: "ns", Value: "admin.$cmd.bulkWrite"}, {Key: "firstBatch", Value: bson.A{item}}}
+	reply := bson.D{
+		{Key: "ok", Value: 1}, {Key: "cursor", Value: cursor}, {Key: "nErrors", Value: 1},
+		{Key: "nInserted", Value: 0}, {Key: "nDeleted", Value: 0}, {Key: "nMatched", Value: 0},
+		{Key: "nModified", Value: 0}, {Key: "nUpserted", Value: 0},
+	}
+	responses := []bson.D{collectionQualificationResponse("db", "records"), reply}
+	writes := 0
+	monitor := &event.CommandMonitor{Started: func(_ context.Context, e *event.CommandStartedEvent) {
+		if e.CommandName == "bulkWrite" {
+			writes++
+		}
+	}}
+	adapter := batchMockAdapter(t, responses, monitor)
+	plans := make([]*execution.Plan, 2)
+	for i := range plans {
+		document := bson.D{{Key: "_id", Value: "x"}, {Key: "n", Value: int32(i)}}
+		opts := batchOperationOptions{resource: "db/records/s:x", action: "put", document: document}
+		work, failure := prepareTestRecord(adapter, batchOperation(t, opts))
+		if failure != nil {
+			t.Fatal(failure)
+		}
+		work.Context = t.Context()
+		plans[i] = work
+	}
+	for i, event := range adapter.ExecuteRecords(t.Context(), plans) {
+		result := event.GetMutationResult()
+		if result.GetOutcome() != pb.MutationOutcome_NOT_APPLIED || result.GetFailure().GetCode() != pb.FailureCode_PERMISSION_DENIED {
+			t.Fatal("put", i, result)
+		}
+	}
+	if writes != 1 {
+		t.Fatal("a permission failure ran the folded puts again one at a time", writes)
+	}
+}
+
+func TestMongoAnAmbiguousChainCommitFailsItsDependentPrograms(t *testing.T) {
+	ambiguousLabels := bson.A{"UnknownTransactionCommitResult"}
+	ambiguous := programCommitError(ambiguousLabels)
+	document := bson.D{{Key: "_id", Value: "x"}, {Key: "n", Value: int32(0)}}
+	cursor := bson.D{{Key: "id", Value: int64(0)}, {Key: "ns", Value: "db.records"}, {Key: "firstBatch", Value: bson.A{document}}}
+	item := bson.D{{Key: "ok", Value: 1}, {Key: "idx", Value: 0}, {Key: "n", Value: 1}, {Key: "nModified", Value: 1}}
+	responses := []bson.D{collectionQualificationResponse("db", "records"), readCursorResponse(cursor), programBulkResponse(bson.A{item}, 1)}
+	for range 5 {
+		responses = append(responses, ambiguous)
+	}
+	adapter := batchMockAdapter(t, responses, nil)
+	programs := [][2]string{{"x", setOneProgram}, {"x", rejectSetProgram}}
+	results := adapter.executePrograms(t.Context(), mockPrograms(t, adapter, programs))
+	first, second := results[0].GetMutationResult(), results[1].GetMutationResult()
+	if first.GetOutcome() != pb.MutationOutcome_UNKNOWN || first.GetFailure().GetCode() != pb.FailureCode_UNAVAILABLE {
+		t.Fatal("the program the write carried", first)
+	}
+	// The second program was rejected on the document only the unknown
+	// write would have made real.
+	if second.GetOutcome() != pb.MutationOutcome_NOT_APPLIED || second.GetFailure().GetCode() != pb.FailureCode_UNAVAILABLE {
+		t.Fatal("the dependent program kept its rejection", second)
+	}
+}
+
+func TestMongoAChainWhoseProgramLeftDuringEvaluationRunsAgainWithoutIt(t *testing.T) {
+	config := Config{Store: "mongo"}
+	adapter := &Adapter{config: config}
+	initializeTestLifetime(adapter)
+	loop := strings.Replace(loopProgramPattern, "%s", strings.Repeat("lowercase", 900), 1)
+	programs := [][2]string{{"x", setOneProgram}, {"x", rejectSetProgram}, {"y", loop}}
+	plans := mockPrograms(t, adapter, programs)
+	// The first program's caller leaves while y's program runs, after x's
+	// chain was evaluated.
+	caller, cancel := context.WithTimeout(t.Context(), 2*time.Second)
+	defer cancel()
+	plans[0].Context, plans[2].Context = caller, caller
+	timer := time.AfterFunc(100*time.Millisecond, cancel)
+	defer timer.Stop()
+	stands, folds := foldPrograms(plans)
+	defer folds.release()
+	batch := &programBatch{plans: stands, results: make([]*pb.MutationResult, len(stands)), folds: folds}
+	var documents []bson.Raw
+	for _, id := range []string{"x", "y"} {
+		document := bson.D{{Key: "_id", Value: id}, {Key: "n", Value: int32(0)}}
+		raw, err := bson.Marshal(document)
+		if err != nil {
+			t.Fatal(err)
+		}
+		documents = append(documents, raw)
+	}
+	positions := []int{0, 1}
+	writes, writePositions := adapter.transformPrograms(t.Context(), batch, positions, documents)
+	if len(writes) != 1 || writePositions[0] != 0 {
+		t.Fatal("writes", writes, writePositions)
+	}
+	native := writes[0].Backend.(*plan)
+	if native.document.Lookup("n").AsInt64() != 0 || native.document.Lookup("m").AsInt64() != 1 {
+		t.Fatal("the departed program's effect was written", native.document)
+	}
+	own := folds.own[stands[0]]
+	if own[0].GetFailure().GetCode() != pb.FailureCode_CANCELLED || own[1] != nil {
+		t.Fatal("chain results", own)
+	}
+}
diff --git a/internal/backend/mongodb/program.go b/internal/backend/mongodb/program.go
index d010b84e..751233fe 100644
--- a/internal/backend/mongodb/program.go
+++ b/internal/backend/mongodb/program.go
@@ -61,6 +61,13 @@ func (a *Adapter) executePrograms(ctx context.Context, plans []*execution.Plan)
 				}
 				byPlan[work] = result
 			}
+			// A chain whose write did not apply also fails the programs
+			// rejected on the document it would have made real.
+			if outcome := batch.results[i].GetOutcome(); outcome != pb.MutationOutcome_APPLIED {
+				for _, j := range folds.dependent[stand] {
+					byPlan[chain[j]] = protocol.Mutation(pb.MutationOutcome_NOT_APPLIED, batch.results[i].GetFailure())
+				}
+			}
 		}
 	}
 	results := make([]*pb.Event, len(plans))
@@ -267,7 +274,8 @@ func (a *Adapter) runProgramAttempt(ctx context.Context, state *programAttempt)
 			for i, reply := range replies {
 				if reply != nil && reply.Outcome == pb.MutationOutcome_NOT_APPLIED {
 					batch.results[writePositions[i]] = reply
-					if stand := batch.plans[writePositions[i]]; batch.folds.chain(stand) != nil {
+					stand := batch.plans[writePositions[i]]
+					if batch.folds.chain(stand) != nil && reply.GetFailure().GetCode() == pb.FailureCode_PRECONDITION_FAILED {
 						batch.folds.unfold[stand] = true
 					}
 				}
@@ -438,25 +446,46 @@ func (a *Adapter) readPrograms(ctx context.Context, plans []*execution.Plan) ([]
 }
 
 func (a *Adapter) transformPrograms(ctx context.Context, batch *programBatch, positions []int, documents []bson.Raw) ([]*execution.Plan, []int) {
-	writes := make([]*execution.Plan, 0, len(positions))
-	writePositions := make([]int, 0, len(positions))
+	writes := make([]*execution.Plan, len(positions))
 	for i, position := range positions {
-		work := batch.plans[position]
-		var write *execution.Plan
-		var result *pb.MutationResult
-		if batch.folds.chain(work) != nil {
-			write, result = a.transformChain(ctx, batch.folds, work, documents[i])
-		} else {
-			write, result = a.transformProgram(ctx, work, documents[i])
-		}
-		if result != nil {
-			batch.results[position] = result
-		} else {
-			writes = append(writes, write)
-			writePositions = append(writePositions, position)
-		}
-	}
-	return writes, writePositions
+		writes[i] = a.transformAt(ctx, batch, position, documents[i])
+	}
+	// A chain whose write carries the effect of a program whose caller left
+	// while others were evaluated is evaluated again without it, on the same
+	// document. Each pass leaves out at least one program.
+	for departed := true; departed; {
+		departed = false
+		for i, position := range positions {
+			if writes[i] != nil && batch.folds.departed(ctx, batch.plans[position]) {
+				departed = true
+				writes[i] = a.transformAt(ctx, batch, position, documents[i])
+			}
+		}
+	}
+	kept := make([]*execution.Plan, 0, len(writes))
+	writePositions := make([]int, 0, len(writes))
+	for i, write := range writes {
+		if write != nil {
+			kept = append(kept, write)
+			writePositions = append(writePositions, positions[i])
+		}
+	}
+	return kept, writePositions
+}
+
+// transformAt evaluates the program or chain at a batch position on its
+// document. It returns the write, or records the result when there is none.
+func (a *Adapter) transformAt(ctx context.Context, batch *programBatch, position int, document bson.Raw) *execution.Plan {
+	work := batch.plans[position]
+	var write *execution.Plan
+	var result *pb.MutationResult
+	if batch.folds.chain(work) != nil {
+		write, result = a.transformChain(ctx, batch.folds, work, document)
+	} else {
+		write, result = a.transformProgram(ctx, work, document)
+	}
+	batch.results[position] = result
+	return write
 }
 
 // transformProgram evaluates one program under its caller's context, which ends

From 1f69ecd8b3f7b1e7442560371b07f891da4560a2 Mon Sep 17 00:00:00 2001
From: Randy <857271370@qq.com>
Date: Sun, 8 Nov 2026 10:43:20 +0800
Subject: [PATCH 3/6] =?UTF-8?q?=E2=9C=85=20test:=20let=20folded=20MongoDB?=
 =?UTF-8?q?=20increments=20share=20a=20commit?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 internal/server/lua_batch_integration_test.go | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/internal/server/lua_batch_integration_test.go b/internal/server/lua_batch_integration_test.go
index fc0b4fc4..41ca545e 100644
--- a/internal/server/lua_batch_integration_test.go
+++ b/internal/server/lua_batch_integration_test.go
@@ -443,10 +443,13 @@ func TestRouteMongoLuaSameURIRPCsPreserveAppliedIncrements(t *testing.T) {
 	for _, observed := range proxy.Events() {
 		counts[observed.Command]++
 	}
-	if counts["find"] < len(cases)+1 || counts["bulkWrite"] < applied || counts["commitTransaction"] != applied {
-		t.Fatal("same-URI Lua RPCs did not use successive transaction snapshots", counts)
+	// Increments that share a batch fold into one transaction write, so each
+	// commit carries one or more of them.
+	commits := counts["commitTransaction"]
+	if applied == 0 || commits < 1 || commits > applied || counts["bulkWrite"] < commits || counts["find"] < commits+1 {
+		t.Fatal("same-URI Lua RPCs did not use transaction snapshots", applied, counts)
 	}
-	t.Log("16 concurrent same-URI Lua RPCs: persisted increments match APPLIED outcomes; document fields unchanged")
+	t.Log("16 concurrent same-URI Lua RPCs: persisted increments match APPLIED outcomes, with at most one commit each; document fields unchanged")
 }
 
 func TestRouteSearchLuaSameURIRPCsPreserveAppliedIncrements(t *testing.T) {

From d06dada90bf75a35854050baf99ea9d709fa8611 Mon Sep 17 00:00:00 2001
From: Randy <857271370@qq.com>
Date: Sun, 8 Nov 2026 11:14:54 +0800
Subject: [PATCH 4/6] =?UTF-8?q?=F0=9F=90=9B=20fix:=20keep=20MongoDB=20no-o?=
 =?UTF-8?q?ps=20decided=20on=20the=20read=20document=20applied?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 internal/backend/mongodb/fold.go      |  8 +++++++-
 internal/backend/mongodb/fold_test.go | 11 ++++++++---
 2 files changed, 15 insertions(+), 4 deletions(-)

diff --git a/internal/backend/mongodb/fold.go b/internal/backend/mongodb/fold.go
index 0a4f2c5c..1e53efe8 100644
--- a/internal/backend/mongodb/fold.go
+++ b/internal/backend/mongodb/fold.go
@@ -139,7 +139,7 @@ func (a *Adapter) transformChain(batch context.Context, folds *programFolds, sta
 	own := make([]*pb.MutationResult, len(chain))
 	folds.own[stand] = own
 	var dependent []int
-	state := raw
+	state, changed := raw, false
 	for i, work := range chain {
 		write, result := a.transformProgram(batch, work, state)
 		switch {
@@ -153,6 +153,12 @@ func (a *Adapter) transformChain(batch context.Context, folds *programFolds, sta
 				dependent = append(dependent, i)
 			}
 		}
+		if own[i] == nil && !changed && sameDocument(raw, state) {
+			// Decided on the document as read and left it unchanged, so the
+			// result does not depend on the chain's write.
+			own[i] = protocol.Mutation(pb.MutationOutcome_APPLIED, nil)
+		}
+		changed = changed || !sameDocument(raw, state)
 	}
 	folds.dependent[stand] = dependent
 	native := stand.Backend.(*plan)
diff --git a/internal/backend/mongodb/fold_test.go b/internal/backend/mongodb/fold_test.go
index 18cf4ede..c2ff3c9d 100644
--- a/internal/backend/mongodb/fold_test.go
+++ b/internal/backend/mongodb/fold_test.go
@@ -73,7 +73,7 @@ func TestMongoFoldedPutsRejectedForAnotherReasonDoNotRunAlone(t *testing.T) {
 	}
 }
 
-func TestMongoAnAmbiguousChainCommitFailsItsDependentPrograms(t *testing.T) {
+func TestMongoAnAmbiguousChainCommitFailsItsDependentProgramsOnly(t *testing.T) {
 	ambiguousLabels := bson.A{"UnknownTransactionCommitResult"}
 	ambiguous := programCommitError(ambiguousLabels)
 	document := bson.D{{Key: "_id", Value: "x"}, {Key: "n", Value: int32(0)}}
@@ -84,9 +84,14 @@ func TestMongoAnAmbiguousChainCommitFailsItsDependentPrograms(t *testing.T) {
 		responses = append(responses, ambiguous)
 	}
 	adapter := batchMockAdapter(t, responses, nil)
-	programs := [][2]string{{"x", setOneProgram}, {"x", rejectSetProgram}}
+	keep := `return function(current, incoming) return weir.keep() end`
+	programs := [][2]string{{"x", keep}, {"x", setOneProgram}, {"x", rejectSetProgram}}
 	results := adapter.executePrograms(t.Context(), mockPrograms(t, adapter, programs))
-	first, second := results[0].GetMutationResult(), results[1].GetMutationResult()
+	// The keep was decided on the document as read, whatever the commit did.
+	if kept := results[0].GetMutationResult(); kept.GetOutcome() != pb.MutationOutcome_APPLIED || kept.GetFailure() != nil {
+		t.Fatal("the keep took the commit's result", kept)
+	}
+	first, second := results[1].GetMutationResult(), results[2].GetMutationResult()
 	if first.GetOutcome() != pb.MutationOutcome_UNKNOWN || first.GetFailure().GetCode() != pb.FailureCode_UNAVAILABLE {
 		t.Fatal("the program the write carried", first)
 	}

From c6487e752af0b7e3f4e0521bb2e05cbd45c9b29b Mon Sep 17 00:00:00 2001
From: Randy <857271370@qq.com>
Date: Sun, 8 Nov 2026 11:52:53 +0800
Subject: [PATCH 5/6] =?UTF-8?q?=F0=9F=90=9B=20fix:=20let=20MongoDB=20check?=
 =?UTF-8?q?=20every=20document=20a=20folded=20Lua=20chain=20writes?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Write one op per program that changes the record, in order, in the
chain's transaction, so validators and unique indexes see each document
as they would one program at a time. Unfold a chain too large for one
command, report NOT_APPLIED rather than NOT_STARTED in a replay, give
every folded put a result when no session starts, and split a write only
when more than one remains after folding.
---
 docs/batching.md                              |  17 ++-
 internal/backend/mongodb/batch.go             |  19 ++-
 internal/backend/mongodb/fold.go              |  42 ++++--
 .../backend/mongodb/fold_integration_test.go  | 131 +++++++++++++++++-
 internal/backend/mongodb/program.go           |  33 +++--
 5 files changed, 205 insertions(+), 37 deletions(-)

diff --git a/docs/batching.md b/docs/batching.md
index f5101a34..8b722a37 100644
--- a/docs/batching.md
+++ b/docs/batching.md
@@ -181,9 +181,9 @@ retain backend concurrency semantics. A batch holds one mutation per resource,
 except where the adapter folds the mutations of one record: those queued from
 one or several streams share a batch in queue order, the adapter applies them in
 that order and sends one write, and each mutation keeps its own result. The
-backend validates only the document the write sends: a document that a later
-mutation replaced before the write is never checked, and the mutation that
-produced it reports APPLIED. Later mutations of a resource do not join a batch
+backend validates only the documents the write sends: where a later mutation
+replaced a document before the write, the backend never checks it, and the
+mutation that produced it reports APPLIED. Later mutations of a resource do not join a batch
 that left out an earlier one. A mutation whose caller leaves before the write
 goes out is left out, and the rest are evaluated again without it. A conflict evaluates them again from a fresh
 read. A rejection of the combined document runs the mutations again one at a
@@ -201,10 +201,13 @@ wait for one refresh, or each for its own when they run one at a time. One whose
 precondition fails or whose Lua rejects leaves the state to the next.
 
 MongoDB folds puts, deletes and Lua programs. A record's puts and deletes become
-the last of them, which leaves the record as the whole run would. Its Lua
-programs evaluate in order on the transaction's one read of the record, each on
-the document the programs before it left, and the transaction writes the result
-once. Creates, replaces and backend expressions do not fold.
+the last of them, which leaves the record as the whole run would; only that
+last document is validated. Its Lua programs evaluate in order on the
+transaction's one read of the record, each on the document the programs before
+it left, and the transaction writes each program's document in order, so
+MongoDB checks every one as it would one program at a time, and one commit
+covers them all. A chain whose documents exceed one command runs one program
+at a time. Creates, replaces and backend expressions do not fold.
 
 Grouping is not an application transaction:
 a MongoDB Lua transaction may include records from several RPCs, and its physical
diff --git a/internal/backend/mongodb/batch.go b/internal/backend/mongodb/batch.go
index 19be008c..ca41badd 100644
--- a/internal/backend/mongodb/batch.go
+++ b/internal/backend/mongodb/batch.go
@@ -301,7 +301,7 @@ func (a *Adapter) executeWrites(ctx context.Context, plans []*execution.Plan) []
 	active, positions, writes, stands := foldWrites(active, positions, writes)
 
 	encoded := encodeBulkWrite(writes, namespaces)
-	if len(plans) > 1 && len(encoded) > a.commandBytes() {
+	if len(active) > 1 && len(encoded) > a.commandBytes() {
 		middle := len(plans) / 2
 		copy(results[:middle], a.executeWrites(ctx, plans[:middle]))
 		copy(results[middle:], a.executeWrites(ctx, plans[middle:]))
@@ -311,8 +311,14 @@ func (a *Adapter) executeWrites(ctx context.Context, plans []*execution.Plan) []
 	state := &writeBatch{plans: active, results: make([]*pb.MutationResult, len(active))}
 	session, err := a.client.StartSession()
 	if err != nil {
-		for i, p := range active {
-			results[positions[i]] = execution.FailedEvent(p.Command, pb.MutationOutcome_NOT_STARTED, backendFailure(ctx, err))
+		for i := range active {
+			members := stands[i]
+			if members == nil {
+				members = positions[i : i+1]
+			}
+			for _, position := range members {
+				results[position] = execution.FailedEvent(plans[position].Command, pb.MutationOutcome_NOT_STARTED, backendFailure(ctx, err))
+			}
 		}
 		return results
 	}
@@ -378,7 +384,12 @@ func (a *Adapter) executeWrites(ctx context.Context, plans []*execution.Plan) []
 			// the chain applied, so each write runs alone.
 			for _, position := range members {
 				single := []*execution.Plan{plans[position]}
-				results[position] = a.executeWrites(base, single)[0]
+				result := a.executeWrites(base, single)[0]
+				if mutation := result.GetMutationResult(); mutation.GetOutcome() == pb.MutationOutcome_NOT_STARTED {
+					// The chain's write already went out.
+					result = execution.FailedEvent(plans[position].Command, pb.MutationOutcome_NOT_APPLIED, mutation.GetFailure())
+				}
+				results[position] = result
 			}
 			continue
 		}
diff --git a/internal/backend/mongodb/fold.go b/internal/backend/mongodb/fold.go
index 1e53efe8..594e6110 100644
--- a/internal/backend/mongodb/fold.go
+++ b/internal/backend/mongodb/fold.go
@@ -131,16 +131,20 @@ func chainContext(chain []*execution.Plan) (context.Context, func()) {
 }
 
 // transformChain evaluates a chain's programs in queue order, each on the
-// document the programs before it left, and returns one write from the read
-// document to the last. A program that does not change the document, or that
-// fails, leaves it to the next.
-func (a *Adapter) transformChain(batch context.Context, folds *programFolds, stand *execution.Plan, raw bson.Raw) (*execution.Plan, *pb.MutationResult) {
+// document the programs before it left, and returns a write for each program
+// that changes the document. The transaction sends them in order, so MongoDB
+// checks every document the chain passes through, as it would one program at
+// a time. A program that does not change the document, or that fails, leaves
+// it to the next.
+func (a *Adapter) transformChain(batch context.Context, folds *programFolds, stand *execution.Plan, raw bson.Raw) ([]*execution.Plan, *pb.MutationResult) {
 	chain := folds.chains[stand]
 	own := make([]*pb.MutationResult, len(chain))
 	folds.own[stand] = own
 	var dependent []int
+	var writes []*execution.Plan
 	state, changed := raw, false
 	for i, work := range chain {
+		previous := state
 		write, result := a.transformProgram(batch, work, state)
 		switch {
 		case result == nil && write.Backend.(*plan).action == "delete":
@@ -159,23 +163,33 @@ func (a *Adapter) transformChain(batch context.Context, folds *programFolds, sta
 			own[i] = protocol.Mutation(pb.MutationOutcome_APPLIED, nil)
 		}
 		changed = changed || !sameDocument(raw, state)
+		if !sameDocument(previous, state) {
+			writes = append(writes, chainWrite(stand, previous, state))
+		}
 	}
 	folds.dependent[stand] = dependent
+	if len(writes) == 0 {
+		unchanged := protocol.Mutation(pb.MutationOutcome_APPLIED, nil)
+		return nil, unchanged
+	}
+	return writes, nil
+}
+
+// chainWrite is the write that takes a chain's record from one document to
+// the next, nil for a missing record.
+func chainWrite(stand *execution.Plan, from, to bson.Raw) *execution.Plan {
 	native := stand.Backend.(*plan)
 	prepared := &plan{target: native.target, id: native.id}
 	switch {
-	case sameDocument(raw, state):
-		unchanged := protocol.Mutation(pb.MutationOutcome_APPLIED, nil)
-		return nil, unchanged
-	case state == nil:
+	case to == nil:
 		prepared.action = "delete"
-	case raw == nil:
-		prepared.action, prepared.document = "create", state
+	case from == nil:
+		prepared.action, prepared.document = "create", to
 	default:
-		prepared.action, prepared.document = "replace", state
+		prepared.action, prepared.document = "replace", to
 	}
 	write := &execution.Plan{Command: stand.Command, Backend: prepared}
-	return write, nil
+	return write
 }
 
 // sameDocument reports whether two documents, nil for a missing one, are
@@ -186,11 +200,13 @@ func sameDocument(a, b bson.Raw) bool {
 }
 
 // unfoldPrograms runs each program of a rejected chain alone, in queue order.
+// The chain's transaction already ran, so a program whose caller left since is
+// NOT_APPLIED rather than NOT_STARTED.
 func (a *Adapter) unfoldPrograms(ctx context.Context, chain []*execution.Plan) []*pb.MutationResult {
 	results := make([]*pb.MutationResult, len(chain))
 	for i, work := range chain {
 		single := []*execution.Plan{work}
-		batch := &programBatch{plans: single, results: make([]*pb.MutationResult, 1)}
+		batch := &programBatch{plans: single, results: make([]*pb.MutationResult, 1), attempted: true}
 		a.runPrograms(ctx, batch)
 		results[i] = batch.results[0]
 	}
diff --git a/internal/backend/mongodb/fold_integration_test.go b/internal/backend/mongodb/fold_integration_test.go
index dcdb98f4..aac93002 100644
--- a/internal/backend/mongodb/fold_integration_test.go
+++ b/internal/backend/mongodb/fold_integration_test.go
@@ -4,6 +4,7 @@ package mongodb
 
 import (
 	"context"
+	"strings"
 	"sync/atomic"
 	"testing"
 
@@ -95,7 +96,7 @@ func storedN(t *testing.T, collection *mongo.Collection, id string) (int64, bool
 	return document.N, true
 }
 
-func TestMongoLuaProgramsOfOneRecordFoldIntoOneTransactionWrite(t *testing.T) {
+func TestMongoLuaProgramsOfOneRecordFoldIntoOneTransaction(t *testing.T) {
 	fixture := testmongo.Open(t)
 	collection := fixture.Admin.Database(fixture.DB).Collection("records")
 	for _, id := range []string{"a", "gone"} {
@@ -122,7 +123,8 @@ func TestMongoLuaProgramsOfOneRecordFoldIntoOneTransactionWrite(t *testing.T) {
 	applied, rejected := pb.MutationOutcome_APPLIED, pb.MutationOutcome_NOT_APPLIED
 	want := []pb.MutationOutcome{applied, applied, applied, applied, rejected, applied, applied, applied}
 	checkOutcomes(t, results, want)
-	if commands.finds.Load() != 1 || commands.writes.Load() != 1 || commands.ops.Load() != 3 || commands.commits.Load() != 1 {
+	// One op for each program that changed its record, in one ordered write.
+	if commands.finds.Load() != 1 || commands.writes.Load() != 1 || commands.ops.Load() != 7 || commands.commits.Load() != 1 {
 		t.Fatal("finds", commands.finds.Load(), "writes", commands.writes.Load(), "ops", commands.ops.Load(), "commits", commands.commits.Load())
 	}
 	if n, found := storedN(t, collection, "a"); !found || n != 4 {
@@ -218,3 +220,128 @@ func TestMongoRejectedFoldedPutsRunOneAtATime(t *testing.T) {
 		t.Fatal("the put before the rejected one did not apply alone", n, found)
 	}
 }
+
+// validatedCollection creates a collection whose documents must keep n at
+// most 10.
+func validatedCollection(t *testing.T, fixture *testmongo.Fixture) *mongo.Collection {
+	t.Helper()
+	database := fixture.Admin.Database(fixture.DB)
+	validator := bson.D{{Key: "n", Value: bson.D{{Key: "$lte", Value: 10}}}}
+	create := options.CreateCollection().SetValidator(validator)
+	if err := database.CreateCollection(t.Context(), "validated", create); err != nil {
+		t.Fatal(err)
+	}
+	collection := database.Collection("validated")
+	return collection
+}
+
+// cancelOnFirstWrite returns a caller context that ends once the first
+// bulkWrite has been answered, and a monitor that ends it.
+func cancelOnFirstWrite(t *testing.T) (context.Context, *event.CommandMonitor) {
+	caller, cancel := context.WithCancel(t.Context())
+	t.Cleanup(cancel)
+	monitor := &event.CommandMonitor{Succeeded: func(_ context.Context, e *event.CommandSucceededEvent) {
+		if e.CommandName == "bulkWrite" {
+			cancel()
+		}
+	}}
+	return caller, monitor
+}
+
+func TestMongoFoldedProgramsMeetEveryCheckAProgramAloneWould(t *testing.T) {
+	fixture := testmongo.Open(t)
+	collection := validatedCollection(t, fixture)
+	document := bson.D{{Key: "_id", Value: "v"}, {Key: "n", Value: int64(1)}}
+	if _, err := collection.InsertOne(t.Context(), document); err != nil {
+		t.Fatal(err)
+	}
+	adapterOpts := adapterTestOptions{fixture: fixture}
+	adapter := testAdapter(t, adapterOpts)
+	records := recordWrites{t: t, adapter: adapter, prefix: fixture.DB + "/validated/s:"}
+	plans := []*execution.Plan{
+		records.program("v", setProgram("50")),
+		records.program("v", `return function(current, incoming) if current.n > 10 then current.n = current.n - 45 end; return current end`),
+	}
+	results := adapter.executePrograms(t.Context(), plans)
+	// Alone, the first program is rejected and the second finds n at 1.
+	want := []pb.MutationOutcome{pb.MutationOutcome_NOT_APPLIED, pb.MutationOutcome_APPLIED}
+	checkOutcomes(t, results, want)
+	if n, found := storedN(t, collection, "v"); !found || n != 1 {
+		t.Fatal("the second program ran on a document MongoDB rejected", n, found)
+	}
+}
+
+func TestMongoAFoldedPutReplacedBeforeTheWriteIsNotChecked(t *testing.T) {
+	fixture := testmongo.Open(t)
+	collection := validatedCollection(t, fixture)
+	adapterOpts := adapterTestOptions{fixture: fixture}
+	adapter := testAdapter(t, adapterOpts)
+	records := recordWrites{t: t, adapter: adapter, prefix: fixture.DB + "/validated/s:"}
+	plans := []*execution.Plan{
+		records.write("v", "put", 50),
+		records.write("v", "put", 1),
+	}
+	// Only the last put reaches MongoDB, so both report APPLIED, as
+	// docs/batching.md states, and the record holds the last.
+	results := adapter.ExecuteRecords(t.Context(), plans)
+	want := []pb.MutationOutcome{pb.MutationOutcome_APPLIED, pb.MutationOutcome_APPLIED}
+	checkOutcomes(t, results, want)
+	if n, found := storedN(t, collection, "v"); !found || n != 1 {
+		t.Fatal("folded puts", n, found)
+	}
+}
+
+func TestMongoAMutationWhoseCallerLeftBeforeItsChainRanAgainIsNotApplied(t *testing.T) {
+	for _, kind := range []string{"put", "program"} {
+		t.Run(kind, func(t *testing.T) {
+			fixture := testmongo.Open(t)
+			validatedCollection(t, fixture)
+			caller, monitor := cancelOnFirstWrite(t)
+			adapterOpts := adapterTestOptions{fixture: fixture, monitor: monitor}
+			adapter := testAdapter(t, adapterOpts)
+			records := recordWrites{t: t, adapter: adapter, prefix: fixture.DB + "/validated/s:"}
+			plans := []*execution.Plan{records.write("v", "put", 1), records.write("v", "put", 50)}
+			execute := adapter.ExecuteRecords
+			if kind == "program" {
+				plans = []*execution.Plan{records.program("v", setProgram("1")), records.program("v", setProgram("50"))}
+				execute = adapter.executePrograms
+			}
+			// The chain is rejected and runs again one at a time, after the
+			// first caller left.
+			plans[0].Context = caller
+			results := execute(t.Context(), plans)
+			want := []pb.MutationOutcome{pb.MutationOutcome_NOT_APPLIED, pb.MutationOutcome_NOT_APPLIED}
+			checkOutcomes(t, results, want)
+			if code := results[0].GetMutationResult().GetFailure().GetCode(); code != pb.FailureCode_CANCELLED {
+				t.Fatal("the departed caller's cause", code)
+			}
+		})
+	}
+}
+
+func TestMongoAChainTooLargeForOneCommandRunsOneAtATime(t *testing.T) {
+	fixture := testmongo.Open(t)
+	collection := fixture.Admin.Database(fixture.DB).Collection("records")
+	commands := &foldCommands{}
+	adapterOpts := adapterTestOptions{fixture: fixture, monitor: commands.monitor()}
+	adapter := testAdapter(t, adapterOpts)
+	settings := adapter.options()
+	settings.ExchangeBytes = 4096
+	adapter.config.Options = &settings
+	padding := strings.Repeat("x", 3000)
+	program := func(n string) string {
+		source := `return function(current, incoming) return {n = ` + n + `, pad = "` + padding + `"} end`
+		return source
+	}
+	records := recordWrites{t: t, adapter: adapter, prefix: fixture.DB + "/records/s:"}
+	plans := []*execution.Plan{
+		records.program("big", program("1")),
+		records.program("big", program("2")),
+	}
+	results := adapter.executePrograms(t.Context(), plans)
+	want := []pb.MutationOutcome{pb.MutationOutcome_APPLIED, pb.MutationOutcome_APPLIED}
+	checkOutcomes(t, results, want)
+	if n, found := storedN(t, collection, "big"); !found || n != 2 || commands.commits.Load() != 2 {
+		t.Fatal("stored", n, found, "commits", commands.commits.Load())
+	}
+}
diff --git a/internal/backend/mongodb/program.go b/internal/backend/mongodb/program.go
index 751233fe..98dc85ea 100644
--- a/internal/backend/mongodb/program.go
+++ b/internal/backend/mongodb/program.go
@@ -240,6 +240,14 @@ func (a *Adapter) runProgramAttempt(ctx context.Context, state *programAttempt)
 		if errors.Is(writeErr, errProgramBatchBound) && len(batch.active()) > 1 {
 			return programAttemptSplit
 		}
+		if remaining := batch.active(); errors.Is(writeErr, errProgramBatchBound) && len(remaining) == 1 && batch.folds.chain(batch.plans[remaining[0]]) != nil {
+			// One chain's documents exceed a command, so its programs run
+			// one at a time.
+			stand := batch.plans[remaining[0]]
+			batch.folds.unfold[stand] = true
+			batch.results[remaining[0]] = protocol.Mutation(pb.MutationOutcome_NOT_APPLIED, nil)
+			return programAttemptFinished
+		}
 		if programHasLabel(writeErr, "TransientTransactionError") && pauseProgram(ctx, a.options().RetryBackoff) {
 			return transientRetry(writeErr)
 		}
@@ -446,7 +454,7 @@ func (a *Adapter) readPrograms(ctx context.Context, plans []*execution.Plan) ([]
 }
 
 func (a *Adapter) transformPrograms(ctx context.Context, batch *programBatch, positions []int, documents []bson.Raw) ([]*execution.Plan, []int) {
-	writes := make([]*execution.Plan, len(positions))
+	writes := make([][]*execution.Plan, len(positions))
 	for i, position := range positions {
 		writes[i] = a.transformAt(ctx, batch, position, documents[i])
 	}
@@ -464,8 +472,8 @@ func (a *Adapter) transformPrograms(ctx context.Context, batch *programBatch, po
 	}
 	kept := make([]*execution.Plan, 0, len(writes))
 	writePositions := make([]int, 0, len(writes))
-	for i, write := range writes {
-		if write != nil {
+	for i, ops := range writes {
+		for _, write := range ops {
 			kept = append(kept, write)
 			writePositions = append(writePositions, positions[i])
 		}
@@ -474,18 +482,21 @@ func (a *Adapter) transformPrograms(ctx context.Context, batch *programBatch, po
 }
 
 // transformAt evaluates the program or chain at a batch position on its
-// document. It returns the write, or records the result when there is none.
-func (a *Adapter) transformAt(ctx context.Context, batch *programBatch, position int, document bson.Raw) *execution.Plan {
+// document. It returns the writes, or records the result when there are none.
+func (a *Adapter) transformAt(ctx context.Context, batch *programBatch, position int, document bson.Raw) []*execution.Plan {
 	work := batch.plans[position]
-	var write *execution.Plan
-	var result *pb.MutationResult
 	if batch.folds.chain(work) != nil {
-		write, result = a.transformChain(ctx, batch.folds, work, document)
-	} else {
-		write, result = a.transformProgram(ctx, work, document)
+		writes, result := a.transformChain(ctx, batch.folds, work, document)
+		batch.results[position] = result
+		return writes
 	}
+	write, result := a.transformProgram(ctx, work, document)
 	batch.results[position] = result
-	return write
+	if write == nil {
+		return nil
+	}
+	writes := []*execution.Plan{write}
+	return writes
 }
 
 // transformProgram evaluates one program under its caller's context, which ends

From 8babb7211551e9983e442c0255126f5d393c03ea Mon Sep 17 00:00:00 2001
From: Randy <857271370@qq.com>
Date: Sun, 8 Nov 2026 12:16:56 +0800
Subject: [PATCH 6/6] =?UTF-8?q?=F0=9F=90=9B=20fix:=20leave=20MongoDB=20pro?=
 =?UTF-8?q?grams=20whose=20callers=20left=20out=20of=20chains?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A program whose caller left before the batch reports NOT_STARTED and
joins no chain, and a chain without a live caller ends at once, so it
never reaches MongoDB. The docs say MongoDB writes each changed document
in one ordered bulkWrite.
---
 docs/batching.md                      |  2 +-
 docs/lua.md                           |  4 ++-
 internal/backend/mongodb/fold.go      |  8 ++++-
 internal/backend/mongodb/fold_test.go | 51 +++++++++++++++++++++++++++
 internal/backend/mongodb/program.go   | 13 +++++--
 5 files changed, 73 insertions(+), 5 deletions(-)

diff --git a/docs/batching.md b/docs/batching.md
index 8b722a37..9bfa5aaa 100644
--- a/docs/batching.md
+++ b/docs/batching.md
@@ -180,7 +180,7 @@ Same-resource mutations within a stream preserve input order. Different streams
 retain backend concurrency semantics. A batch holds one mutation per resource,
 except where the adapter folds the mutations of one record: those queued from
 one or several streams share a batch in queue order, the adapter applies them in
-that order and sends one write, and each mutation keeps its own result. The
+that order in one write request, and each mutation keeps its own result. The
 backend validates only the documents the write sends: where a later mutation
 replaced a document before the write, the backend never checks it, and the
 mutation that produced it reports APPLIED. Later mutations of a resource do not join a batch
diff --git a/docs/lua.md b/docs/lua.md
index 58097b05..a624140e 100644
--- a/docs/lua.md
+++ b/docs/lua.md
@@ -141,7 +141,9 @@ Confirmed MongoDB transaction and Search compare-and-swap conflicts can trigger
 a bounded fresh read and reevaluation. Search evaluates the queued mutations of
 one record in queue order on one read and writes once, so each Lua callback sees
 the result of the mutations before it, and a conflict reevaluates them all.
-MongoDB does the same for the Lua programs of one record in its transaction.
+MongoDB evaluates the Lua programs of one record the same way in its
+transaction and writes each changed document in one ordered bulkWrite and one
+commit.
 Ambiguous write or commit outcomes never
 replay the business callback. A callback error or invalid document does not
 write.
diff --git a/internal/backend/mongodb/fold.go b/internal/backend/mongodb/fold.go
index 594e6110..cf0e6226 100644
--- a/internal/backend/mongodb/fold.go
+++ b/internal/backend/mongodb/fold.go
@@ -111,9 +111,12 @@ func chainContext(chain []*execution.Plan) (context.Context, func()) {
 		ctx, cancel = context.WithDeadline(context.Background(), latest)
 	}
 	var live atomic.Int64
-	live.Store(int64(len(chain)))
 	stops := make([]func() bool, 0, len(chain))
 	for _, work := range chain {
+		if work.Context.Err() != nil {
+			continue
+		}
+		live.Add(1)
 		stop := context.AfterFunc(work.Context, func() {
 			if live.Add(-1) == 0 {
 				cancel()
@@ -121,6 +124,9 @@ func chainContext(chain []*execution.Plan) (context.Context, func()) {
 		})
 		stops = append(stops, stop)
 	}
+	if len(stops) == 0 {
+		cancel()
+	}
 	release := func() {
 		for _, stop := range stops {
 			stop()
diff --git a/internal/backend/mongodb/fold_test.go b/internal/backend/mongodb/fold_test.go
index c2ff3c9d..6a6f9e8c 100644
--- a/internal/backend/mongodb/fold_test.go
+++ b/internal/backend/mongodb/fold_test.go
@@ -142,3 +142,54 @@ func TestMongoAChainWhoseProgramLeftDuringEvaluationRunsAgainWithoutIt(t *testin
 		t.Fatal("chain results", own)
 	}
 }
+
+func TestMongoProgramsWhoseCallersLeftBeforeTheBatchDoNotStart(t *testing.T) {
+	document := bson.D{{Key: "_id", Value: "x"}, {Key: "n", Value: int32(0)}}
+	cursor := bson.D{{Key: "id", Value: int64(0)}, {Key: "ns", Value: "db.records"}, {Key: "firstBatch", Value: bson.A{document}}}
+	item := bson.D{{Key: "ok", Value: 1}, {Key: "idx", Value: 0}, {Key: "n", Value: 1}, {Key: "nModified", Value: 1}}
+	items := bson.A{item}
+	success := bson.D{{Key: "ok", Value: 1}}
+	responses := []bson.D{collectionQualificationResponse("db", "records"), readCursorResponse(cursor), programBulkResponse(items, 1), success}
+	var commands []string
+	monitor := &event.CommandMonitor{Started: func(_ context.Context, e *event.CommandStartedEvent) {
+		commands = append(commands, e.CommandName)
+	}}
+	adapter := batchMockAdapter(t, responses, monitor)
+	programs := [][2]string{{"x", setOneProgram}, {"x", setOneProgram}, {"y", setOneProgram}, {"y", setOneProgram}}
+	plans := mockPrograms(t, adapter, programs)
+	left, cancel := context.WithCancel(t.Context())
+	cancel()
+	// Every caller of y left, and one of x.
+	plans[0].Context, plans[2].Context, plans[3].Context = left, left, left
+	results := adapter.executePrograms(t.Context(), plans)
+	for _, i := range []int{0, 2, 3} {
+		result := results[i].GetMutationResult()
+		if result.GetOutcome() != pb.MutationOutcome_NOT_STARTED || result.GetFailure().GetCode() != pb.FailureCode_CANCELLED {
+			t.Fatal("program", i, result)
+		}
+	}
+	if result := results[1].GetMutationResult(); result.GetOutcome() != pb.MutationOutcome_APPLIED {
+		t.Fatal("the remaining program", result)
+	}
+	// Only x is read: y has no caller left.
+	finds := 0
+	for _, name := range commands {
+		if name == "find" {
+			finds++
+		}
+	}
+	if finds != 1 {
+		t.Fatal("the departed chain reached MongoDB", commands)
+	}
+}
+
+func TestAChainWhoseCallersAllLeftEndsAtOnce(t *testing.T) {
+	left, cancel := context.WithCancel(t.Context())
+	cancel()
+	chain := []*execution.Plan{{Context: left}, {Context: left}}
+	ctx, release := chainContext(chain)
+	defer release()
+	if ctx.Err() == nil {
+		t.Fatal("a chain without a remaining caller is live")
+	}
+}
diff --git a/internal/backend/mongodb/program.go b/internal/backend/mongodb/program.go
index 98dc85ea..0ec7ddbb 100644
--- a/internal/backend/mongodb/program.go
+++ b/internal/backend/mongodb/program.go
@@ -36,13 +36,22 @@ type programBatch struct {
 // order, and one write. Caller contexts govern evaluation; the batch context
 // governs I/O.
 func (a *Adapter) executePrograms(ctx context.Context, plans []*execution.Plan) []*pb.Event {
-	stands, folds := foldPrograms(plans)
+	// A program whose caller already left neither starts nor joins a chain.
+	byPlan := make(map[*execution.Plan]*pb.MutationResult, len(plans))
+	live := make([]*execution.Plan, 0, len(plans))
+	for _, work := range plans {
+		if failure := work.Ended(ctx); failure != nil {
+			byPlan[work] = protocol.Mutation(pb.MutationOutcome_NOT_STARTED, failure)
+			continue
+		}
+		live = append(live, work)
+	}
+	stands, folds := foldPrograms(live)
 	if folds != nil {
 		defer folds.release()
 	}
 	batch := &programBatch{plans: stands, results: make([]*pb.MutationResult, len(stands)), folds: folds}
 	a.runPrograms(ctx, batch)
-	byPlan := make(map[*execution.Plan]*pb.MutationResult, len(plans))
 	for i, stand := range stands {
 		chain := folds.chain(stand)
 		switch {