diff --git a/docs/batching.md b/docs/batching.md index 7eac33e..9bfa5aa 100644 --- a/docs/batching.md +++ b/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. diff --git a/docs/lua.md b/docs/lua.md index 32bebdd..a624140 100644 --- a/docs/lua.md +++ b/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. diff --git a/docs/operations.md b/docs/operations.md index 7d10423..2bed681 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 199d035..cae98c3 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 0754f89..ca41bad 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,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 } diff --git a/internal/backend/mongodb/fold.go b/internal/backend/mongodb/fold.go new file mode 100644 index 0000000..cf0e622 --- /dev/null +++ b/internal/backend/mongodb/fold.go @@ -0,0 +1,273 @@ +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. + // 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() +} + +func (f *programFolds) chain(stand *execution.Plan) []*execution.Plan { + if f == nil { + return nil + } + 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() + } +} + +// 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), + dependent: make(map[*execution.Plan][]int), + 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 + 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() + } + }) + stops = append(stops, stop) + } + if len(stops) == 0 { + cancel() + } + 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 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": + state = nil + case result == nil: + 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) + } + } + 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) + 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 to == nil: + prepared.action = "delete" + case from == nil: + prepared.action, prepared.document = "create", to + default: + prepared.action, prepared.document = "replace", to + } + write := &execution.Plan{Command: stand.Command, Backend: prepared} + return write +} + +// 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. +// 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), attempted: true} + 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 0000000..aac9300 --- /dev/null +++ b/internal/backend/mongodb/fold_integration_test.go @@ -0,0 +1,347 @@ +//go:build integration + +package mongodb + +import ( + "context" + "strings" + "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 TestMongoLuaProgramsOfOneRecordFoldIntoOneTransaction(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) + // 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 { + 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) + } +} + +// 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/fold_test.go b/internal/backend/mongodb/fold_test.go new file mode 100644 index 0000000..6a6f9e8 --- /dev/null +++ b/internal/backend/mongodb/fold_test.go @@ -0,0 +1,195 @@ +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 TestMongoAnAmbiguousChainCommitFailsItsDependentProgramsOnly(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) + 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)) + // 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) + } + // 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) + } +} + +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 298d7fa..0ec7ddb 100644 --- a/internal/backend/mongodb/program.go +++ b/internal/backend/mongodb/program.go @@ -28,17 +28,60 @@ 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))} + // 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) + 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 + } + // 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)) - 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 } @@ -206,6 +249,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) } @@ -240,6 +291,10 @@ 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 + stand := batch.plans[writePositions[i]] + if batch.folds.chain(stand) != nil && reply.GetFailure().GetCode() == pb.FailureCode_PRECONDITION_FAILED { + batch.folds.unfold[stand] = true + } } } if len(batch.active()) == len(active) { @@ -324,7 +379,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] @@ -408,19 +463,49 @@ 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] - write, result := a.transformProgram(ctx, work, documents[i]) - if result != nil { - batch.results[position] = result - } else { - writes = append(writes, write) - writePositions = append(writePositions, position) + 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]) + } } } - return writes, writePositions + kept := make([]*execution.Plan, 0, len(writes)) + writePositions := make([]int, 0, len(writes)) + for i, ops := range writes { + for _, write := range ops { + 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 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] + if batch.folds.chain(work) != nil { + 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 + if write == nil { + return nil + } + writes := []*execution.Plan{write} + return writes } // transformProgram evaluates one program under its caller's context, which ends diff --git a/internal/server/lua_batch_integration_test.go b/internal/server/lua_batch_integration_test.go index fc0b4fc..41ca545 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) {