Fold mutations of one MongoDB record into one write

docs/batching.md

@@ -180,10 +180,10 @@ 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
-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
+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
 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
@@ -200,6 +200,15 @@ 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; 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
 membership varies with arrivals and available capacity.

docs/lua.md

@@ -140,6 +140,10 @@ 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 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.

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

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()

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,9 +297,11 @@ 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() {
+	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:]))
@@ -307,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
 	}
@@ -330,10 +340,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 +378,29 @@ 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 && 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]}
+				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
+		}
+		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
 }