Fold mutations of one Search record into one write

docs/architecture.md

@@ -48,7 +48,8 @@ not full run below an adaptive limit that halves when executions queue at the
 backend, so records wait in Weir and form larger batches instead. See
 [batching](batching.md) for grouping, the adaptive limit and backpressure.
 
-Same-resource mutations within a stream execute in input order. Each plan keeps
+Same-resource mutations within a stream execute in input order: one at a time,
+or together in one batch where the adapter folds them. Each plan keeps
 its own record context, bounded by the record's timeout, its stream and any
 cancel request naming it, and each result returns to the stream that submitted
 it under its ordinal, so equal ordinals in different RPCs cannot collide. A

docs/batching.md

@@ -177,7 +177,30 @@ target. One larger legal item proceeds alone. Full-source
 Lua replacements can generate much more traffic than small incoming patches.
 
 Same-resource mutations within a stream preserve input order. Different streams
-retain backend concurrency semantics. Grouping is not an application transaction:
+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 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
+time, since none applied. Any other failed or unknown write gives its result to
+every mutation it carried, and fails, as not applied, every mutation rejected on
+a state that only that write would have made real. A mutation decided on the
+record as read that leaves it unchanged is applied whatever the write's result.
+`weir_store_folded_mutations_total` counts mutations that followed an earlier
+one of the same resource in their batch.
+
+Search folds every mutation but backend expressions, applying them in order to
+the record's state. The write is conditional on a real-time read unless the
+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.
+
+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

@@ -138,6 +138,8 @@ memory pressure still rejects new work, and deployment memory limits remain the
 final process boundary.
 
 Confirmed MongoDB transaction and Search compare-and-swap conflicts can trigger
-a bounded fresh read and reevaluation. Ambiguous write or commit outcomes never
+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
 replay the business callback. A callback error or invalid document does not
 write.

docs/operations.md

@@ -97,6 +97,7 @@ of improved capacity.
 | `weir_store_native_completions_total` | Final Native forwarding terminal evidence (`not_started`, `response_complete`, `response_incomplete`), not backend business success, RPC status, or client receipt |
 | `weir_store_rejections_total` | Preparation, draining and overload denials, not completed records |
 | `weir_store_batch_operations`, `weir_store_queue_wait_seconds`, `weir_store_execution_seconds` | Record batch sizes, queue wait and record batch execution time; Natives are excluded |
+| `weir_store_folded_mutations_total` | Mutations dispatched in a batch after an earlier mutation of the same resource, for the adapter to [fold](batching.md) into one write; counted at dispatch, including those the adapter later runs one at a time or that lost their caller |
 | `weir_lua_evaluation_seconds` | Lua evaluation time by `outcome` (`success`, `error`, `timeout`, `canceled`), for both backends |
 | `weir_ingress_sessions`, `weir_rpc_completions_total`, `weir_admission_rejections_total` | Active RPCs, completed RPCs by `status` and RPCs refused at admission |
 | `weir_watchdog_expirations_total` | RPCs ended by the input or output stall timeout |
@@ -253,7 +254,9 @@ 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, so a client that keeps rewriting the same records queues behind itself.
+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
+itself.
 
 ### WeirWatchdogExpirations
 

internal/app/diagnostics_integration_test.go

@@ -48,10 +48,10 @@ func TestDiagnosticsMaximumStaticSeries(t *testing.T) {
 			delete(families, name)
 		}
 	}
-	// 16 Mongo Stores (62 runtime series each), plus the business transport
+	// 16 Mongo Stores (63 runtime series each), plus the business transport
 	// (11), Node readiness (1) and the memory guard (3). Histograms count
 	// every bucket, sum and count.
-	const maximumSeries = 16*62 + 11 + 1 + 3
+	const maximumSeries = 16*63 + 11 + 1 + 3
 	if got := testmetrics.Series(families); got != maximumSeries {
 		t.Fatal("maximum static series changed", got)
 	}

internal/backend/search/adapter.go

@@ -55,6 +55,9 @@ type plan struct {
 	program          *luaengine.Program
 	input            *pb.Document
 	waitUntilVisible bool
+	// conditional marks a write framed from a read, which a concurrent
+	// change turns into a conflict that permits evaluating it again.
+	conditional bool
 }
 
 type capabilities struct {
@@ -261,6 +264,9 @@ func (a *Adapter) PrepareRecord(ctx context.Context, record *execution.Record) (
 		native.waitUntilVisible, work.Waits = true, true
 		work.BatchKey += "|visible"
 	}
+	// Every action but a backend expression can be applied to a known source,
+	// so mutations of one record fold into one write.
+	work.Folds = true
 	switch action := mutation.Action.(type) {
 	case *pb.MutateRequest_Put:
 		native.action = "index"
@@ -287,6 +293,7 @@ func (a *Adapter) PrepareRecord(ctx context.Context, record *execution.Record) (
 		expression := action.AtomicTransform.GetBackendExpression()
 		native.action = "expression"
 		native.source = expression.Data
+		work.Folds = false
 		return work, validateExpression(expression)
 	}
 	return nil, protocol.Fail(pb.FailureCode_UNSUPPORTED, "mutation unsupported")

internal/backend/search/batch.go

@@ -180,6 +180,17 @@ type recordBatch struct {
 	pending  []recordWrite
 	retry    []int
 	barriers map[string][]int
+	// chains maps the first position of each chain of folding mutations of
+	// one record to all its positions, in queue order. applied maps a chain
+	// whose write is framed to the positions whose effects that write
+	// carries, and dependent to those rejected on a state that only that
+	// write makes real. unfold lists chains whose combined document was
+	// rejected: nothing of them applied, so their mutations run again one at
+	// a time.
+	chains    map[int][]int
+	applied   map[int][]int
+	dependent map[int][]int
+	unfold    []int
 }
 
 // fail records a final result for one work.
@@ -192,6 +203,12 @@ func (b *recordBatch) fail(position int, outcome pb.MutationOutcome, failure *pb
 func (b *recordBatch) appendWrite(work *execution.Plan, position int, current *getReply) {
 	entry, err := appendBulkEntry(b.entry[:0], work.Backend.(*plan), current)
 	if err != nil {
+		if b.applied[position] != nil {
+			delete(b.applied, position)
+			delete(b.dependent, position)
+			b.unfold = append(b.unfold, position)
+			return
+		}
 		b.fail(position, pb.MutationOutcome_NOT_APPLIED, protocol.Fail(pb.FailureCode_INVALID_ARGUMENT, "Search source cannot be framed as one NDJSON line"))
 		return
 	}
@@ -265,7 +282,18 @@ func (b *recordBatch) flush() {
 	works := make([]*execution.Plan, 0, len(b.pending))
 	positions := make([]int, 0, len(b.pending))
 	for _, item := range b.pending {
-		if failure := item.work.Ended(b.ctx); failure != nil {
+		if members := b.applied[item.position]; members != nil {
+			if b.departed(members) {
+				// The write carries the effect of a mutation whose caller
+				// left, so the chain is evaluated again without it. That is
+				// not a conflict attempt.
+				delete(b.applied, item.position)
+				delete(b.dependent, item.position)
+				b.attempts[item.position]--
+				b.retry = append(b.retry, item.position)
+				continue
+			}
+		} else if failure := item.work.Ended(b.ctx); failure != nil {
 			b.fail(item.position, pb.MutationOutcome_NOT_APPLIED, failure)
 			continue
 		}
@@ -289,10 +317,14 @@ func (b *recordBatch) flush() {
 			b.queueBulkVisibilityBarriers(works, positions, replies, items)
 		}
 		for i, position := range positions {
+			if b.applied[position] != nil {
+				b.settleChain(position, replies[i])
+				continue
+			}
 			b.results[position] = replies[i]
-			program := works[i].Backend.(*plan).program
+			conditional := works[i].Backend.(*plan).conditional
 			mutation := replies[i].GetMutationResult()
-			if program == nil || mutation.GetOutcome() != pb.MutationOutcome_NOT_APPLIED || mutation.GetFailure().GetCode() != pb.FailureCode_CONFLICT {
+			if !conditional || mutation.GetOutcome() != pb.MutationOutcome_NOT_APPLIED || mutation.GetFailure().GetCode() != pb.FailureCode_CONFLICT {
 				continue
 			}
 			if b.attempts[position] < b.adapter.options().RetryAttempts {
@@ -398,15 +430,17 @@ func (b *recordBatch) read(positions []int) {
 
 // mutate frames each work's write, pre-reading the records its action depends
 // on, and sends conflicting transforms around again from a fresh read.
+// Folding mutations of one record become one write.
 func (b *recordBatch) mutate(positions []int) {
 	settings := b.adapter.options()
 	// At most this many pre-read sources are retained while Lua runs.
 	retained := max(1, settings.ExchangeBytes/settings.BatchBytes)
+	positions = b.chain(positions)
 	for len(positions) != 0 {
 		for start := 0; start < len(positions); {
 			end, reads := start, 0
-			for end < len(positions) && (reads < retained || !b.works[positions[end]].Backend.(*plan).preReads()) {
-				if b.works[positions[end]].Backend.(*plan).preReads() {
+			for end < len(positions) && (reads < retained || !b.preReads(positions[end])) {
+				if b.preReads(positions[end]) {
 					reads++
 				}
 				end++
@@ -419,9 +453,78 @@ func (b *recordBatch) mutate(positions []int) {
 		positions = b.retry
 		b.retry = nil
 		for _, i := range positions {
+			b.clear(i)
+		}
+	}
+	for len(b.unfold) != 0 {
+		head := b.unfold[0]
+		b.unfold = b.unfold[1:]
+		members := b.chains[head]
+		delete(b.chains, head)
+		for _, i := range members {
 			b.results[i] = nil
+			b.attempts[i] = 0
+		}
+		for _, i := range members {
+			single := []int{i}
+			b.mutate(single)
+		}
+	}
+}
+
+// chain links folding mutations of one record into chains and returns the
+// positions of the other mutations and of each chain's first mutation.
+func (b *recordBatch) chain(positions []int) []int {
+	starts := make([]int, 0, len(positions))
+	first := make(map[string]int, len(positions))
+	for _, i := range positions {
+		work := b.works[i]
+		head, linked := first[work.Key]
+		if !work.Folds || !linked {
+			if work.Folds {
+				first[work.Key] = i
+			}
+			starts = append(starts, i)
+			continue
+		}
+		if b.chains == nil {
+			b.chains = make(map[int][]int)
+		}
+		if b.chains[head] == nil {
+			b.chains[head] = []int{head}
+		}
+		b.chains[head] = append(b.chains[head], i)
+	}
+	return starts
+}
+
+// clear forgets the results of a work, or of every work of its chain.
+func (b *recordBatch) clear(i int) {
+	members := b.chains[i]
+	if members == nil {
+		b.results[i] = nil
+		return
+	}
+	for _, member := range members {
+		b.results[member] = nil
+	}
+}
+
+// preReads reports whether a work, or a chain starting at it, depends on the
+// record's current state. A chain does unless its first live mutation is a put
+// or delete, which sets the record without reading it.
+func (b *recordBatch) preReads(i int) bool {
+	members := b.chains[i]
+	if members == nil {
+		return b.works[i].Backend.(*plan).preReads()
+	}
+	for _, member := range members {
+		if b.works[member].Ended(b.ctx) == nil {
+			action := b.works[member].Backend.(*plan).action
+			return action != "index" && action != "delete"
 		}
 	}
+	return false
 }
 
 // preReads reports whether a mutation depends on the record's current state.
@@ -432,8 +535,22 @@ func (native *plan) preReads() bool {
 func (b *recordBatch) mutateChunk(chunk []int) {
 	var readWorks []*execution.Plan
 	var readPositions []int
+	// readChains holds the chains that asked for a read; one without its
+	// read has failed or waits for the next round.
+	var readChains map[int]bool
 	for _, i := range chunk {
 		work := b.works[i]
+		if b.chains[i] != nil {
+			if b.preReads(i) {
+				if readChains == nil {
+					readChains = make(map[int]bool)
+				}
+				readChains[i] = true
+				readWorks = append(readWorks, b.reader(i))
+				readPositions = append(readPositions, i)
+			}
+			continue
+		}
 		if denied := work.Ended(b.ctx); denied != nil {
 			b.fail(i, pb.MutationOutcome_NOT_APPLIED, denied)
 			continue
@@ -445,16 +562,37 @@ func (b *recordBatch) mutateChunk(chunk []int) {
 	}
 	current := make(map[int]*getReply, len(readWorks))
 	for offset, record := range b.adapter.mget(b.ctx, readWorks) {
-		if record.failure != nil {
-			b.fail(readPositions[offset], pb.MutationOutcome_NOT_APPLIED, record.failure)
-		} else {
-			current[readPositions[offset]] = record.reply
+		position := readPositions[offset]
+		switch {
+		case record.failure == nil:
+			current[position] = record.reply
+		case b.chains[position] != nil && readWorks[offset].Context.Err() != nil && b.live(position):
+			// The reading caller left; another one reads in the next round.
+			b.retry = append(b.retry, position)
+		case b.chains[position] != nil:
+			for _, member := range b.chains[position] {
+				failure := record.failure
+				if denied := b.works[member].Ended(b.ctx); denied != nil {
+					failure = denied
+				}
+				b.fail(member, pb.MutationOutcome_NOT_APPLIED, failure)
+			}
+		default:
+			b.fail(position, pb.MutationOutcome_NOT_APPLIED, record.failure)
 		}
 	}
 	for _, i := range chunk {
 		if b.results[i] != nil {
 			continue
 		}
+		if b.chains[i] != nil {
+			if readChains[i] && current[i] == nil {
+				continue
+			}
+			b.mutateChain(i, current[i])
+			delete(current, i)
+			continue
+		}
 		if denied := b.works[i].Ended(b.ctx); denied != nil {
 			b.fail(i, pb.MutationOutcome_NOT_APPLIED, denied)
 			continue

internal/backend/search/bulk.go

@@ -126,11 +126,11 @@ func (a *Adapter) bulkResults(works []*execution.Plan, status int, raw []byte, e
 	for i, work := range works {
 		native := work.Backend.(*plan)
 		item := &reply.items[i]
-		transformed := native.program != nil || native.action == "expression"
+		transformed := native.conditional || native.action == "expression"
 		acknowledgement := "write"
 		if transformed {
 			acknowledgement = "update"
-			if native.program != nil {
+			if native.conditional {
 				acknowledgement = "conditional write"
 			}
 			failure := protocol.Fail(pb.FailureCode_UNAVAILABLE, acknowledgement+" acknowledgement unavailable or incomplete")
@@ -145,7 +145,7 @@ func (a *Adapter) bulkResults(works []*execution.Plan, status int, raw []byte, e
 			if status < 400 || failure == nil || item.result != "" || item.version.set || item.seq.set || item.term.set || item.hasShards {
 				continue
 			}
-			if native.action == "create" && native.program == nil && failure.Code == pb.FailureCode_CONFLICT {
+			if native.action == "create" && !native.conditional && failure.Code == pb.FailureCode_CONFLICT {
 				failure = protocol.Fail(pb.FailureCode_PRECONDITION_FAILED, "record already exists")
 			}
 			results[i] = execution.FailedEvent(work.Command, pb.MutationOutcome_NOT_APPLIED, failure)
@@ -157,12 +157,12 @@ func (a *Adapter) bulkResults(works []*execution.Plan, status int, raw []byte, e
 			valid = item.status == 201 && item.result == "created"
 		case "index":
 			valid = item.status == 200 && item.result == "updated" ||
-				native.program == nil && item.status == 201 && item.result == "created"
+				!native.conditional && item.status == 201 && item.result == "created"
 		case "replace":
 			valid = item.status == 200 && item.result == "updated"
 		case "delete":
 			valid = item.status == 200 && item.result == "deleted" ||
-				native.program == nil && item.status == 404 && item.result == "not_found"
+				!native.conditional && item.status == 404 && item.result == "not_found"
 		case "expression":
 			valid = item.status == 200 && (item.result == "updated" || item.result == "noop")
 		}