From f0e1b9cbd9ff7b68fbaaf4a75d4fe87b8b9c58df Mon Sep 17 00:00:00 2001 From: Randy <857271370@qq.com> Date: Sun, 8 Nov 2026 10:15:33 +0800 Subject: [PATCH 1/3] =?UTF-8?q?=E2=9C=A8=20feat:=20fold=20mutations=20of?= =?UTF-8?q?=20one=20Search=20record=20into=20one=20write?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A batch held at most one mutation per resource, so a client that sends several mutations of one record, such as one seller merged once per offer of a search page, waited for each one in turn, and query-visible writes waited for one refresh each. A mutation now says whether its adapter can fold it, and folding mutations of one resource share a batch in queue order, from one stream or several, while a stream's order still holds. Search folds every action but backend expressions. It applies a record's mutations in order to the state of one real-time read, or of none when the first one is a put or delete, and sends one write, conditional on that read. Each mutation keeps its own result: a failed precondition or Lua rejection leaves the state to the next, a conflict evaluates the chain again from a fresh read, and any other rejection runs the mutations one at a time, since none applied. weir_store_folded_mutations_total counts folded mutations. --- docs/architecture.md | 3 +- docs/batching.md | 13 +- docs/lua.md | 4 +- docs/operations.md | 5 +- internal/backend/search/adapter.go | 7 + internal/backend/search/batch.go | 147 ++++++- internal/backend/search/bulk.go | 10 +- internal/backend/search/fold.go | 173 +++++++++ .../backend/search/fold_integration_test.go | 63 +++ internal/backend/search/fold_test.go | 364 ++++++++++++++++++ internal/backend/search/program.go | 1 + internal/backend/search/program_reply_test.go | 12 +- .../backend/search/request_contract_test.go | 2 +- internal/backend/search/visibility.go | 6 + internal/execution/adapter.go | 8 +- internal/store/metrics.go | 5 +- internal/store/runtime.go | 69 +++- internal/store/runtime_test.go | 143 +++++++ 18 files changed, 988 insertions(+), 47 deletions(-) create mode 100644 internal/backend/search/fold.go create mode 100644 internal/backend/search/fold_integration_test.go create mode 100644 internal/backend/search/fold_test.go diff --git a/docs/architecture.md b/docs/architecture.md index ee1cf65b..f8479666 100644 --- a/docs/architecture.md +++ b/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 diff --git a/docs/batching.md b/docs/batching.md index eea03ca2..58d892f7 100644 --- a/docs/batching.md +++ b/docs/batching.md @@ -177,7 +177,18 @@ 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 that Search folds the mutations of one record, apart from backend +expressions: those queued from one or several streams share a batch in queue +order, and Search applies them in that order to the record's state and sends one +write. The write is conditional on a real-time read unless the first mutation is +a put or delete, and query-visible mutations wait for one refresh. Each mutation +keeps its own result. One whose precondition fails or whose Lua rejects leaves +the state to the next. A conflict evaluates the chain again from a fresh read. +Any other rejection of the combined write runs the mutations again one at a +time, since none applied; an unknown write leaves every mutation it carried +unknown. `weir_store_folded_mutations_total` counts mutations that followed an +earlier one of the same resource in their batch. 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 400f8bef..32bebddc 100644 --- a/docs/lua.md +++ b/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. diff --git a/docs/operations.md b/docs/operations.md index 96a7cfc4..3841ee49 100644 --- a/docs/operations.md +++ b/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, which the adapter [folds](batching.md) into one write | | `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 diff --git a/internal/backend/search/adapter.go b/internal/backend/search/adapter.go index 30ad322f..ce773d2c 100644 --- a/internal/backend/search/adapter.go +++ b/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") diff --git a/internal/backend/search/batch.go b/internal/backend/search/batch.go index 613ea0f3..cfe00735 100644 --- a/internal/backend/search/batch.go +++ b/internal/backend/search/batch.go @@ -180,6 +180,14 @@ 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. unfold lists chains whose write was rejected: nothing of them + // applied, so their mutations run again one at a time. + chains map[int][]int + applied map[int][]int + unfold []int } // fail records a final result for one work. @@ -192,6 +200,11 @@ 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) + 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 +278,13 @@ 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 { + // A chain's write goes out while any of its callers remains. + if b.endChain(members) { + delete(b.applied, 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 +308,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 +421,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 +444,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 +// replaces or deletes the record whatever it holds. +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 +526,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 +553,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 diff --git a/internal/backend/search/bulk.go b/internal/backend/search/bulk.go index 8f65ddc0..56a6d5f7 100644 --- a/internal/backend/search/bulk.go +++ b/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") } diff --git a/internal/backend/search/fold.go b/internal/backend/search/fold.go new file mode 100644 index 00000000..5f39fbe1 --- /dev/null +++ b/internal/backend/search/fold.go @@ -0,0 +1,173 @@ +package search + +import ( + "bytes" + + "github.com/batchstream/weir-protocol/api/protocol" + pb "github.com/batchstream/weir-protocol/api/weir/v1" + "github.com/batchstream/weir/internal/execution" +) + +// reader returns the work that reads a chain's record: its first mutation +// whose caller remains, so that one caller leaving does not fail the read. +func (b *recordBatch) reader(head int) *execution.Plan { + for _, member := range b.chains[head] { + if b.works[member].Ended(b.ctx) == nil { + return b.works[member] + } + } + return b.works[head] +} + +// live reports whether any caller of a chain remains. +func (b *recordBatch) live(head int) bool { + for _, member := range b.chains[head] { + if b.works[member].Ended(b.ctx) == nil { + return true + } + } + return false +} + +// mutateChain applies a chain's mutations in queue order to the record's +// state and frames one write that takes the record from the state it read, or +// from any state when the first mutation replaces or deletes it regardless, +// to the last. A mutation that is rejected leaves the state to the next one. +func (b *recordBatch) mutateChain(head int, current *getReply) { + if current == nil && b.preReads(head) { + // The mutation that made the read unnecessary lost its caller. + b.retry = append(b.retry, head) + return + } + b.attempts[head]++ + found, source := false, []byte(nil) + if current != nil { + found, source = *current.Found, current.Source + } + var applied []int + for _, i := range b.chains[head] { + work := b.works[i] + if denied := work.Ended(b.ctx); denied != nil { + b.fail(i, pb.MutationOutcome_NOT_APPLIED, denied) + continue + } + step := work.Backend.(*plan) + var rejected *pb.MutationResult + switch step.action { + case "index": + found, source = true, step.source + case "delete": + found, source = false, nil + case "create": + if found { + rejected = protocol.Mutation(pb.MutationOutcome_NOT_APPLIED, protocol.Fail(pb.FailureCode_PRECONDITION_FAILED, "record already exists")) + break + } + found, source = true, step.source + case "replace": + if !found { + rejected = protocol.Mutation(pb.MutationOutcome_NOT_APPLIED, protocol.Fail(pb.FailureCode_PRECONDITION_FAILED, "record missing")) + break + } + source = step.source + case "program": + present := found + state := &getReply{Found: &present, Source: source} + next, mutation := b.adapter.evaluateProgram(work, state) + if denied := work.Ended(b.ctx); denied != nil { + mutation = protocol.Mutation(pb.MutationOutcome_NOT_APPLIED, denied) + } + switch { + case mutation == nil && next.action == "delete": + found, source = false, nil + case mutation == nil: + found, source = true, next.source + case mutation.Outcome != pb.MutationOutcome_APPLIED || mutation.Failure != nil: + rejected = mutation + } + } + if rejected != nil { + b.fail(i, rejected.Outcome, rejected.Failure) + continue + } + applied = append(applied, i) + } + if len(applied) == 0 { + return + } + native := b.works[head].Backend.(*plan) + read := current != nil + var action string + switch { + case read && !*current.Found && !found: + case read && *current.Found && found && bytes.Equal(source, current.Source): + case !found: + action = "delete" + case read && !*current.Found: + action = "create" + default: + action = "index" + } + if action == "" { + // The record already holds the chain's result. + for _, i := range applied { + b.fail(i, pb.MutationOutcome_APPLIED, nil) + if native.waitUntilVisible { + b.queueVisibilityBarrier(native.index, i) + } + } + return + } + write := &plan{index: native.index, id: native.id, action: action, waitUntilVisible: native.waitUntilVisible, conditional: read} + if found { + write.source = source + } + framed := *b.works[head] + framed.Backend = write + if b.applied == nil { + b.applied = make(map[int][]int) + } + b.applied[head] = applied + b.appendWrite(&framed, head, current) +} + +// endChain fails a framed chain whose callers all left and reports whether +// it did. +func (b *recordBatch) endChain(members []int) bool { + for _, i := range members { + if b.works[i].Ended(b.ctx) == nil { + return false + } + } + for _, i := range members { + b.fail(i, pb.MutationOutcome_NOT_APPLIED, b.works[i].Ended(b.ctx)) + } + return true +} + +// settleChain gives a chain's write result to each mutation the write +// carried. A conflict evaluates the chain again from a fresh read, and any +// other rejection runs its mutations one at a time, since none applied. +func (b *recordBatch) settleChain(head int, reply *pb.Event) { + members := b.applied[head] + delete(b.applied, head) + mutation := reply.GetMutationResult() + if mutation.GetOutcome() == pb.MutationOutcome_NOT_APPLIED { + if mutation.GetFailure().GetCode() != pb.FailureCode_CONFLICT { + b.unfold = append(b.unfold, head) + return + } + if b.attempts[head] < b.adapter.options().RetryAttempts { + b.retry = append(b.retry, head) + return + } + failure := protocol.Fail(pb.FailureCode_CONFLICT, "Search record changed during every transform attempt") + for _, i := range members { + b.fail(i, pb.MutationOutcome_NOT_APPLIED, failure) + } + return + } + for _, i := range members { + b.fail(i, mutation.GetOutcome(), mutation.GetFailure()) + } +} diff --git a/internal/backend/search/fold_integration_test.go b/internal/backend/search/fold_integration_test.go new file mode 100644 index 00000000..ce61d970 --- /dev/null +++ b/internal/backend/search/fold_integration_test.go @@ -0,0 +1,63 @@ +//go:build integration + +package search + +import ( + "net/http" + "testing" + + pb "github.com/batchstream/weir-protocol/api/weir/v1" + "github.com/batchstream/weir/internal/execution" +) + +func TestSearchFoldedMutationsOfOneRecordApplyAsOneVisibleWrite(t *testing.T) { + adapter, backend := setupSearch(t) + status, raw := backend.Do(t, "PUT", "/"+backend.Index+"/_settings", `{"index":{"refresh_interval":"100ms"}}`) + if status != http.StatusOK { + t.Fatal("fixture refresh interval", status, string(raw)) + } + status, raw = backend.Do(t, "PUT", "/"+backend.Index+"/_doc/seller?refresh=true", `{"n":1}`) + if status != http.StatusCreated { + t.Fatal("fixture document", status, string(raw)) + } + seller := recordMutations{t: t, adapter: adapter, resource: searchResource(backend.Index, "seller"), visible: true} + other := recordMutations{t: t, adapter: adapter, resource: searchResource(backend.Index, "other"), visible: true} + fresh := recordMutations{t: t, adapter: adapter, resource: searchResource(backend.Index, "fresh"), visible: true} + precondition := expected{outcome: pb.MutationOutcome_NOT_APPLIED, code: pb.FailureCode_PRECONDITION_FAILED} + + works := []*execution.Plan{ + seller.prepare("add", `{"by":1}`), + other.prepare("put", `{"n":7}`), + seller.prepare("add", `{"by":10}`), + seller.prepare("add", `{"by":100}`), + } + want := []expected{appliedResult, appliedResult, appliedResult, appliedResult} + checkResults(t, adapter.ExecuteRecords(t.Context(), works), want) + sellerState := searchVisibilityState{id: "seller", n: 112} + assertVisibleSearchState(t, backend, sellerState) + otherState := searchVisibilityState{id: "other", n: 7} + assertVisibleSearchState(t, backend, otherState) + + works = []*execution.Plan{ + fresh.prepare("create", `{"n":1}`), + fresh.prepare("create", `{"n":2}`), + fresh.prepare("replace", `{"n":3}`), + fresh.prepare("delete", ``), + fresh.prepare("replace", `{"n":4}`), + fresh.prepare("put", `{"n":5}`), + fresh.prepare("add", `{"by":4}`), + } + want = []expected{appliedResult, precondition, appliedResult, appliedResult, precondition, appliedResult, appliedResult} + checkResults(t, adapter.ExecuteRecords(t.Context(), works), want) + freshState := searchVisibilityState{id: "fresh", n: 9} + assertVisibleSearchState(t, backend, freshState) + + works = []*execution.Plan{ + seller.prepare("add", `{"by":1}`), + seller.prepare("delete", ``), + } + want = []expected{appliedResult, appliedResult} + checkResults(t, adapter.ExecuteRecords(t.Context(), works), want) + deleted := searchVisibilityState{id: "seller", missing: true} + assertVisibleSearchState(t, backend, deleted) +} diff --git a/internal/backend/search/fold_test.go b/internal/backend/search/fold_test.go new file mode 100644 index 00000000..df30246a --- /dev/null +++ b/internal/backend/search/fold_test.go @@ -0,0 +1,364 @@ +package search + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "slices" + "strings" + "sync" + "testing" + + pb "github.com/batchstream/weir-protocol/api/weir/v1" + "github.com/batchstream/weir/internal/execution" +) + +type storedSource struct { + source string + seq int64 +} + +// memorySearch serves one index from memory with real-time reads and the +// conditional bulk writes Weir sends. +type memorySearch struct { + mu sync.Mutex + docs map[string]storedSource + seq int64 + reads int + bulks []string + refreshes []string + barriers int + // changeBeforeWrite rewrites a stored record before the next bulk applies, + // as a concurrent writer would. + changeBeforeWrite []string + reject func(source string) bool +} + +func (m *memorySearch) handle(w http.ResponseWriter, r *http.Request) { + m.mu.Lock() + defer m.mu.Unlock() + switch r.URL.Path { + case "/records": + _, _ = io.WriteString(w, testIndexReply) + case "/records/_mget": + m.reads++ + var request struct{ IDs []string } + _ = json.NewDecoder(r.Body).Decode(&request) + docs := make([]string, len(request.IDs)) + for i, id := range request.IDs { + doc, found := m.docs[id] + if !found { + docs[i] = fmt.Sprintf(`{"_index":"records","_id":%q,"found":false}`, id) + continue + } + docs[i] = fmt.Sprintf(`{"_index":"records","_id":%q,"found":true,"_seq_no":%d,"_primary_term":1,"_source":%s}`, id, doc.seq, doc.source) + } + _, _ = io.WriteString(w, `{"docs":[`+strings.Join(docs, ",")+`]}`) + case "/records/_refresh": + m.barriers++ + _, _ = io.WriteString(w, `{"_shards":{"total":1,"successful":1,"failed":0}}`) + case "/_bulk": + raw, _ := io.ReadAll(r.Body) + m.bulks = append(m.bulks, string(raw)) + m.refreshes = append(m.refreshes, r.URL.Query().Get("refresh")) + for _, id := range m.changeBeforeWrite { + m.seq++ + m.docs[id] = storedSource{source: `{"n":100}`, seq: m.seq} + } + m.changeBeforeWrite = nil + items, failed := m.apply(string(raw)) + _, _ = fmt.Fprintf(w, `{"errors":%t,"took":1,"items":[%s]}`, failed, strings.Join(items, ",")) + default: + http.Error(w, "unexpected path", http.StatusNotFound) + } +} + +func (m *memorySearch) apply(body string) ([]string, bool) { + var items []string + failed := false + scanner := bufio.NewScanner(strings.NewReader(body)) + for scanner.Scan() { + var line map[string]struct { + ID string `json:"_id"` + Seq *int64 `json:"if_seq_no"` + Term *int64 `json:"if_primary_term"` + } + _ = json.Unmarshal(scanner.Bytes(), &line) + for op, meta := range line { + source := "" + if op != "delete" && scanner.Scan() { + source = scanner.Text() + } + doc, found := m.docs[meta.ID] + reply := func(status int, fields string) string { + return fmt.Sprintf(`{%q:{"_index":"records","_id":%q,"status":%d,%s}}`, op, meta.ID, status, fields) + } + switch { + case meta.Seq != nil && (!found || doc.seq != *meta.Seq), op == "create" && found: + items, failed = append(items, reply(409, `"error":{"type":"version_conflict_engine_exception"}`)), true + continue + case m.reject != nil && op != "delete" && m.reject(source): + items, failed = append(items, reply(400, `"error":{"type":"mapper_parsing_exception"}`)), true + continue + } + m.seq++ + status, result := 200, "updated" + switch { + case op == "delete" && found: + result = "deleted" + delete(m.docs, meta.ID) + case op == "delete": + status, result = 404, "not_found" + case !found: + status, result = 201, "created" + m.docs[meta.ID] = storedSource{source: source, seq: m.seq} + default: + m.docs[meta.ID] = storedSource{source: source, seq: m.seq} + } + fields := fmt.Sprintf(`"result":%q,"_version":2,"_seq_no":%d,"_primary_term":1,"_shards":{"total":1,"successful":1,"failed":0}`, result, m.seq) + items = append(items, reply(status, fields)) + } + } + return items, failed +} + +func (m *memorySearch) source(id string) string { + m.mu.Lock() + defer m.mu.Unlock() + return m.docs[id].source +} + +func newMemorySearch(t *testing.T, docs map[string]string) (*memorySearch, *Adapter) { + t.Helper() + memory := &memorySearch{docs: make(map[string]storedSource)} + for id, source := range docs { + memory.seq++ + memory.docs[id] = storedSource{source: source, seq: memory.seq} + } + adapter := testHTTPAdapter(t, memory.handle) + return memory, adapter +} + +const addProgram = `return function(current, incoming) + local n = 0 + if current ~= nil then n = current.n end + return {n = n + incoming.by} +end` + +// recordMutations prepares mutations of one resource. +type recordMutations struct { + t *testing.T + adapter *Adapter + resource string + visible bool +} + +// prepare prepares one mutation. data is the document of a put, create or +// replace, or the input of a Lua program. +func (r recordMutations) prepare(action, data string) *execution.Plan { + r.t.Helper() + request := &pb.MutateRequest{Resource: r.resource} + document := &pb.Document{ContentType: "application/json", Data: []byte(data)} + switch action { + case "put": + request.Action = &pb.MutateRequest_Put{Put: document} + case "create": + request.Action = &pb.MutateRequest_Create{Create: document} + case "replace": + request.Action = &pb.MutateRequest_Replace{Replace: document} + case "delete": + empty := &pb.Empty{} + request.Action = &pb.MutateRequest_Delete{Delete: empty} + case "add", "keep": + source := addProgram + if action == "keep" { + source = `return function(current, incoming) return weir.keep() end` + } + lua := &pb.LuaTransform{Source: []byte(source), Input: document} + form := &pb.Transform_Lua{Lua: lua} + transform := &pb.Transform{Form: form} + request.Action = &pb.MutateRequest_AtomicTransform{AtomicTransform: transform} + default: + r.t.Fatal(action) + } + if r.visible { + request.Options = &pb.MutationOptions{Visibility: pb.MutationVisibility_QUERY_VISIBLE} + } + variant := &pb.Command_Mutate{Mutate: request} + command := &pb.Command{Operation: variant} + operation := &pb.ExecuteRequest{Index: 1, Command: command} + work, failure := prepareTestRecord(r.adapter, operation) + if failure != nil { + r.t.Fatal(failure) + } + return work +} + +type expected struct { + outcome pb.MutationOutcome + code pb.FailureCode +} + +var appliedResult = expected{outcome: pb.MutationOutcome_APPLIED} + +func checkResults(t *testing.T, results []*pb.Event, want []expected) { + t.Helper() + if len(results) != len(want) { + t.Fatal("results", len(results), "want", len(want)) + } + for i, result := range results { + mutation := result.GetMutationResult() + if mutation.GetOutcome() != want[i].outcome || mutation.GetFailure().GetCode() != want[i].code { + t.Fatalf("result %d: %v, want %v", i, mutation, want[i]) + } + } +} + +func TestOneRecordsMutationsFoldIntoOneReadAndOneWrite(t *testing.T) { + memory, adapter := newMemorySearch(t, map[string]string{"seller": `{"n":1}`}) + seller := recordMutations{t: t, adapter: adapter, resource: "records/s:seller", visible: true} + other := recordMutations{t: t, adapter: adapter, resource: "records/s:other", visible: true} + works := []*execution.Plan{ + seller.prepare("add", `{"by":1}`), + other.prepare("put", `{"n":7}`), + seller.prepare("add", `{"by":10}`), + seller.prepare("add", `{"by":100}`), + } + results := adapter.ExecuteRecords(t.Context(), works) + want := []expected{appliedResult, appliedResult, appliedResult, appliedResult} + checkResults(t, results, want) + if memory.reads != 1 || len(memory.bulks) != 1 || memory.refreshes[0] != "wait_for" { + t.Fatal("reads", memory.reads, "bulks", len(memory.bulks), memory.refreshes) + } + if lines := strings.Count(memory.bulks[0], "\n"); lines != 4 { + t.Fatal("one write per record expected", memory.bulks[0]) + } + if !strings.Contains(memory.bulks[0], `"if_seq_no":1`) { + t.Fatal("the folded write was not conditional on its read", memory.bulks[0]) + } + if got := memory.source("seller"); got != `{"n":112}` { + t.Fatal("folded transforms did not apply in order", got) + } +} + +func TestFoldedActionsApplyInQueueOrder(t *testing.T) { + memory, adapter := newMemorySearch(t, map[string]string{"x": `{"n":1}`}) + x := recordMutations{t: t, adapter: adapter, resource: "records/s:x"} + works := []*execution.Plan{ + x.prepare("put", `{"a":1}`), + x.prepare("create", `{"b":1}`), + x.prepare("replace", `{"n":5}`), + x.prepare("add", `{"by":1}`), + x.prepare("delete", ``), + x.prepare("replace", `{"d":1}`), + x.prepare("create", `{"n":40}`), + x.prepare("add", `{"by":2}`), + } + results := adapter.ExecuteRecords(t.Context(), works) + precondition := expected{outcome: pb.MutationOutcome_NOT_APPLIED, code: pb.FailureCode_PRECONDITION_FAILED} + want := []expected{appliedResult, precondition, appliedResult, appliedResult, appliedResult, precondition, appliedResult, appliedResult} + checkResults(t, results, want) + // A put that replaces whatever the record holds needs no read. + if memory.reads != 0 || len(memory.bulks) != 1 || strings.Contains(memory.bulks[0], "if_seq_no") { + t.Fatal("reads", memory.reads, memory.bulks) + } + if got := memory.source("x"); got != `{"n":42}` { + t.Fatal("folded actions", got) + } +} + +func TestAFoldedWriteThatConflictsRunsAgainFromAFreshRead(t *testing.T) { + memory, adapter := newMemorySearch(t, map[string]string{"x": `{"n":1}`}) + memory.changeBeforeWrite = []string{"x"} + x := recordMutations{t: t, adapter: adapter, resource: "records/s:x"} + works := []*execution.Plan{ + x.prepare("add", `{"by":1}`), + x.prepare("add", `{"by":2}`), + } + results := adapter.ExecuteRecords(t.Context(), works) + want := []expected{appliedResult, appliedResult} + checkResults(t, results, want) + if memory.reads != 2 || len(memory.bulks) != 2 { + t.Fatal("reads", memory.reads, "bulks", len(memory.bulks)) + } + if got := memory.source("x"); got != `{"n":103}` { + t.Fatal("the chain did not run again on the concurrent change", got) + } +} + +func TestARejectedFoldedWriteRunsEachMutationAlone(t *testing.T) { + memory, adapter := newMemorySearch(t, nil) + memory.reject = func(source string) bool { return strings.Contains(source, "bad") } + x := recordMutations{t: t, adapter: adapter, resource: "records/s:x"} + works := []*execution.Plan{ + x.prepare("put", `{"n":1}`), + x.prepare("put", `{"bad":1}`), + x.prepare("add", `{"by":1}`), + } + results := adapter.ExecuteRecords(t.Context(), works) + rejected := expected{outcome: pb.MutationOutcome_NOT_APPLIED, code: pb.FailureCode_PRECONDITION_FAILED} + // Alone, the transform reads the rejected put's predecessor, whose n it + // increments; folded, it would have failed with the rejected document. + want := []expected{appliedResult, rejected, appliedResult} + checkResults(t, results, want) + if got := memory.source("x"); got != `{"n":2}` { + t.Fatal("mutations did not run alone after the rejection", got) + } + if len(memory.bulks) != 4 { + t.Fatal("one folded write and three single writes expected", len(memory.bulks)) + } +} + +func TestAFoldedChainWithoutChangeWritesNothing(t *testing.T) { + memory, adapter := newMemorySearch(t, map[string]string{"x": `{"n":1}`}) + x := recordMutations{t: t, adapter: adapter, resource: "records/s:x", visible: true} + works := []*execution.Plan{ + x.prepare("keep", `{}`), + x.prepare("keep", `{}`), + } + results := adapter.ExecuteRecords(t.Context(), works) + want := []expected{appliedResult, appliedResult} + checkResults(t, results, want) + if len(memory.bulks) != 0 || memory.barriers != 1 { + t.Fatal("bulks", len(memory.bulks), "refresh barriers", memory.barriers) + } +} + +func TestAChainMemberWhoseCallerLeftDoesNotStart(t *testing.T) { + memory, adapter := newMemorySearch(t, map[string]string{"x": `{"n":1}`}) + x := recordMutations{t: t, adapter: adapter, resource: "records/s:x"} + works := []*execution.Plan{ + x.prepare("add", `{"by":1}`), + x.prepare("add", `{"by":10}`), + x.prepare("add", `{"by":100}`), + } + ended, cancel := context.WithCancel(t.Context()) + cancel() + works[0].Context = ended + results := adapter.ExecuteRecords(t.Context(), works) + canceled := expected{outcome: pb.MutationOutcome_NOT_STARTED, code: pb.FailureCode_CANCELLED} + want := []expected{canceled, appliedResult, appliedResult} + checkResults(t, results, want) + if got := memory.source("x"); got != `{"n":111}` { + t.Fatal("the remaining chain did not apply", got) + } +} + +func TestBackendExpressionsDoNotFold(t *testing.T) { + adapter := &Adapter{dialect: ElasticsearchProduct} + expression := batchTestPlan(t, adapter, "expression", "records/s:x") + if expression.Folds { + t.Fatal("a backend expression cannot be applied to a known source") + } + for _, action := range []string{"put", "create", "replace", "delete", "program"} { + if work := batchTestPlan(t, adapter, action, "records/s:x"); !work.Folds { + t.Fatal(action, "does not fold") + } + } + if read := batchTestPlan(t, adapter, "read", "records/s:x"); read.Folds || !slices.Contains([]string{"records"}, read.BatchKey) { + t.Fatal("a read folds") + } +} diff --git a/internal/backend/search/program.go b/internal/backend/search/program.go index e210ea43..f9242e19 100644 --- a/internal/backend/search/program.go +++ b/internal/backend/search/program.go @@ -41,6 +41,7 @@ func (a *Adapter) evaluateProgram(work *execution.Plan, current *getReply) (*pla return nil, execution.LuaFailure(ctx, err) } next := *native + next.conditional = true switch transformed.Action { case "keep": return nil, protocol.Mutation(pb.MutationOutcome_APPLIED, nil) diff --git a/internal/backend/search/program_reply_test.go b/internal/backend/search/program_reply_test.go index 8a97cac7..aea7f946 100644 --- a/internal/backend/search/program_reply_test.go +++ b/internal/backend/search/program_reply_test.go @@ -36,7 +36,7 @@ func TestProgramWriteReplyRecognizesRejectionEnvelopes(t *testing.T) { t.Run(tc.product+"/"+tc.kind, func(t *testing.T) { a := &Adapter{dialect: tc.product} program := &luaengine.Program{} - native := &plan{index: "records", id: "item", action: "index", program: program} + native := &plan{index: "records", id: "item", action: "index", program: program, conditional: true} raw := fmt.Sprintf(`{"_index":"records","_id":"item","error":{"type":%q},"status":%d}`, tc.kind, tc.status) result := testBulkMutationReply(a, native, raw) if result.GetOutcome() != pb.MutationOutcome_NOT_APPLIED || result.GetFailure().GetCode() != tc.failure { @@ -47,7 +47,7 @@ func TestProgramWriteReplyRecognizesRejectionEnvelopes(t *testing.T) { // Lua create conflicts must retain CONFLICT so the caller can reevaluate. a := &Adapter{dialect: ElasticsearchProduct} program := &luaengine.Program{} - native := &plan{index: "records", id: "item", action: "create", program: program} + native := &plan{index: "records", id: "item", action: "create", program: program, conditional: true} raw := `{"_index":"records","_id":"item","error":{"type":"version_conflict_engine_exception"},"status":409}` result := testBulkMutationReply(a, native, raw) if result.GetOutcome() != pb.MutationOutcome_NOT_APPLIED || result.GetFailure().GetCode() != pb.FailureCode_CONFLICT { @@ -68,7 +68,7 @@ func TestProgramWriteReplyRequiresCompleteSuccessEvidence(t *testing.T) { } for _, tc := range cases { program := &luaengine.Program{} - native := &plan{index: "records", id: "item", action: tc.operation, program: program} + native := &plan{index: "records", id: "item", action: tc.operation, program: program, conditional: true} raw := strings.Replace(success, `"updated"`, fmt.Sprintf("%q", tc.result), 1) raw = strings.Replace(raw, `"status":200`, fmt.Sprintf(`"status":%d`, tc.status), 1) result := testBulkMutationReply(a, native, raw) @@ -95,7 +95,7 @@ func TestProgramWriteReplyRequiresCompleteSuccessEvidence(t *testing.T) { `{"_index":"records","_id":"item","error":{"type":"mapper_parsing_exception"},"status":200,"result":"updated"}`, } program := &luaengine.Program{} - native := &plan{index: "records", id: "item", action: "index", program: program} + native := &plan{index: "records", id: "item", action: "index", program: program, conditional: true} for _, raw := range invalid { result := testBulkMutationReply(a, native, raw) if result.GetOutcome() != pb.MutationOutcome_UNKNOWN { @@ -112,7 +112,7 @@ func TestProgramWriteReplyRequiresCompleteSuccessEvidence(t *testing.T) { func TestProgramWriteReplyRejectsContradictoryErrorEvidence(t *testing.T) { a := &Adapter{dialect: ElasticsearchProduct} program := &luaengine.Program{} - native := &plan{index: "records", id: "item", action: "index", program: program} + native := &plan{index: "records", id: "item", action: "index", program: program, conditional: true} for _, field := range []string{`"result":"updated"`, `"_version":1`, `"_seq_no":1`, `"_primary_term":1`, `"_shards":{}`} { raw := fmt.Sprintf(`{"_index":"records","_id":"item","error":{"type":"mapper_parsing_exception"},"status":400,%s}`, field) result := testBulkMutationReply(a, native, raw) @@ -161,7 +161,7 @@ func TestBulkValidatesAllItemsBeforeAcknowledgingAny(t *testing.T) { a := &Adapter{dialect: ElasticsearchProduct} ordinary := &plan{index: "records", id: "item", action: "index"} program := &luaengine.Program{} - native := &plan{index: "records", id: "item", action: "index", program: program} + native := &plan{index: "records", id: "item", action: "index", program: program, conditional: true} ordinaryWork := &execution.Plan{Backend: ordinary} programWork := &execution.Plan{Backend: native} success := `{"_index":"records","_id":"item","status":200,"_version":1,"_seq_no":0,"_primary_term":1,"result":"updated","_shards":{"total":1,"successful":1,"failed":0}}` diff --git a/internal/backend/search/request_contract_test.go b/internal/backend/search/request_contract_test.go index 09bd6d61..1e4314e6 100644 --- a/internal/backend/search/request_contract_test.go +++ b/internal/backend/search/request_contract_test.go @@ -36,7 +36,7 @@ func TestSearchLuaBulkRequiresPositivePrimaryEvidence(t *testing.T) { total, failed = 2, 1 } program := &luaengine.Program{} - native := &plan{index: "records", id: "item", action: operation, program: program} + native := &plan{index: "records", id: "item", action: operation, program: program, conditional: true} empty := &pb.Empty{} mutationAction := &pb.MutateRequest_Delete{Delete: empty} mutation := &pb.MutateRequest{Resource: "records/s:item", Action: mutationAction} diff --git a/internal/backend/search/visibility.go b/internal/backend/search/visibility.go index 2953528a..b0ce61af 100644 --- a/internal/backend/search/visibility.go +++ b/internal/backend/search/visibility.go @@ -11,10 +11,16 @@ import ( "github.com/batchstream/weir/internal/execution" ) +// queueVisibilityBarrier queues a barrier for a work, or for every mutation a +// framed chain's write carries. func (b *recordBatch) queueVisibilityBarrier(index string, position int) { if b.barriers == nil { b.barriers = make(map[string][]int) } + if members := b.applied[position]; members != nil { + b.barriers[index] = append(b.barriers[index], members...) + return + } b.barriers[index] = append(b.barriers[index], position) } diff --git a/internal/execution/adapter.go b/internal/execution/adapter.go index 4bbd8823..508a0a2c 100644 --- a/internal/execution/adapter.go +++ b/internal/execution/adapter.go @@ -23,6 +23,10 @@ type Plan struct { // than the backend's work on it, such as a scheduled refresh. Its time // does not show how loaded the backend is. Waits bool + // Folds marks a mutation the adapter can combine with other folding + // mutations of the same resource in one batch. Such a batch holds them in + // queue order, and the adapter applies them as if one after another. + Folds bool // Context carries this caller's cancellation and deadline inside a shared // batch. The Store runtime replaces the one PrepareRecord set. Context context.Context @@ -55,7 +59,9 @@ type Adapter interface { PrepareRecord(context.Context, *Record) (*Plan, *pb.Failure) // ExecuteRecords runs one batch and returns exactly one result event per // plan, in order. Every plan of a batch shares its BatchKey and is either a - // read or a mutation, and mutations of a batch name distinct resources. + // read or a mutation. Mutations of a batch name distinct resources, except + // that folding mutations of one resource may appear together, in queue + // order. ExecuteRecords(context.Context, []*Plan) []*pb.Event // ExecuteNative streams the response head and chunks through Emit and // returns the terminal NativeEnd. It performs no backend I/O afterwards. diff --git a/internal/store/metrics.go b/internal/store/metrics.go index 75fd05f7..0e3d9afc 100644 --- a/internal/store/metrics.go +++ b/internal/store/metrics.go @@ -8,6 +8,7 @@ import ( type runtimeMetrics struct { records, failures, nativeCompletions, rejections *prometheus.CounterVec queueWait, duration, batch, native prometheus.Histogram + folded prometheus.Counter } func newRuntimeMetrics() runtimeMetrics { @@ -18,6 +19,7 @@ func newRuntimeMetrics() runtimeMetrics { queueOptions := prometheus.HistogramOpts{Name: "weir_store_queue_wait_seconds", Help: "Record time from queueing to dispatch.", Buckets: []float64{.001, .01, .1, 1, 10}} durationOptions := prometheus.HistogramOpts{Name: "weir_store_execution_seconds", Help: "Record batch execution time.", Buckets: []float64{.001, .01, .1, 1, 10}} nativeDurationOptions := prometheus.HistogramOpts{Name: "weir_store_native_seconds", Help: "Native call time, including sending the response on the RPC stream.", Buckets: []float64{.001, .01, .1, 1, 10}} + foldedOptions := prometheus.CounterOpts{Name: "weir_store_folded_mutations_total", Help: "Mutations dispatched in a batch after an earlier mutation of the same resource, which the adapter applies with it."} batchOptions := prometheus.HistogramOpts{Name: "weir_store_batch_operations", Help: "Records per record batch.", Buckets: prometheus.ExponentialBuckets(1, 2, 11)} metrics := runtimeMetrics{ records: prometheus.NewCounterVec(recordOptions, []string{"operation", "outcome"}), @@ -28,6 +30,7 @@ func newRuntimeMetrics() runtimeMetrics { duration: prometheus.NewHistogram(durationOptions), batch: prometheus.NewHistogram(batchOptions), native: prometheus.NewHistogram(nativeDurationOptions), + folded: prometheus.NewCounter(foldedOptions), } for _, outcome := range []string{"applied", "not_applied", "not_started", "unknown"} { metrics.records.WithLabelValues("mutate", outcome) @@ -132,7 +135,7 @@ func (r *Runtime) Collect(ch chan<- prometheus.Metric) { if collector, ok := r.adapter.(prometheus.Collector); ok { collector.Collect(ch) } - collectors := []prometheus.Collector{r.metrics.records, r.metrics.failures, r.metrics.nativeCompletions, r.metrics.rejections, r.metrics.queueWait, r.metrics.duration, r.metrics.batch, r.metrics.native} + collectors := []prometheus.Collector{r.metrics.records, r.metrics.failures, r.metrics.nativeCompletions, r.metrics.rejections, r.metrics.queueWait, r.metrics.duration, r.metrics.batch, r.metrics.native, r.metrics.folded} for _, collector := range collectors { collector.Collect(ch) } diff --git a/internal/store/runtime.go b/internal/store/runtime.go index 20e4cc89..b1decd80 100644 --- a/internal/store/runtime.go +++ b/internal/store/runtime.go @@ -69,8 +69,10 @@ type Runtime struct { limit [2]executionLimit probes probes // probeNext and fullNext describe the batch about to launch: the limit's - // probe, or one that is full. + // probe, or one that is full. foldedNext counts its mutations that follow + // an earlier one of the same resource. probeNext, fullNext bool + foldedNext int draining, closed, overloaded bool // space and nativeSpace are closed and replaced whenever a waiting record // or Native submitter may proceed. @@ -85,7 +87,7 @@ type Runtime struct { } // Session identifies one record stream. Its mutations of one resource execute -// one at a time, in submission order. +// in submission order: one at a time, or together in one batch when they fold. type Session struct { runtime *Runtime // results, when set, receives each record's final event as it completes. @@ -382,15 +384,21 @@ type group struct { items []*Ticket bytes int full bool - // Mutations of one resource never share a batch. - resources map[string]struct{} + // resources records each mutated resource and whether its mutations so + // far all fold. Mutations of one resource share a batch only when they + // all fold. + resources map[string]bool + folded int } type selection struct { - groups []*group - used int - byKey map[groupKey]*group - ordered map[resourceKey]struct{} + groups []*group + used int + byKey map[groupKey]*group + // ordered maps a session's resource to the group its last scanned + // mutation joined, or to nil once a mutation was held back, which holds + // back its later ones too. + ordered map[resourceKey]*group } // reset empties the scratch state after a scan so it retains no tickets; it @@ -399,7 +407,7 @@ func (s *selection) reset() { for _, g := range s.groups[:s.used] { clear(g.items) g.items = g.items[:0] - g.bytes, g.full = 0, false + g.bytes, g.full, g.folded = 0, false, 0 clear(g.resources) } s.used = 0 @@ -431,37 +439,54 @@ func (r *Runtime) selectLocked(now time.Time) ([]*Ticket, time.Time) { s := &r.selection if s.byKey == nil { s.byKey = make(map[groupKey]*group) - s.ordered = make(map[resourceKey]struct{}) + s.ordered = make(map[resourceKey]*group) } defer s.reset() for _, t := range r.queue { plan := t.plan mutation := plan.Command.GetMutate() != nil - if t.session != nil && mutation { + key := groupKey{batch: plan.BatchKey, read: plan.Command.GetRead() != nil} + ordered := t.session != nil && mutation + var resource resourceKey + if ordered { // The oldest queued mutation of a session's resource runs first, - // and only after the previous one finished. - key := resourceKey{session: t.session, resource: plan.Key} - if _, waiting := s.ordered[key]; waiting || r.keys[key] != nil { + // and only after the previous one finished. Later ones join its + // batch while each folds with those before it. + resource = resourceKey{session: t.session, resource: plan.Key} + last, scanned := s.ordered[resource] + if r.keys[resource] != nil || scanned && (last == nil || s.byKey[key] != last) { + s.ordered[resource] = nil continue } - s.ordered[key] = struct{}{} } - key := groupKey{batch: plan.BatchKey, read: plan.Command.GetRead() != nil} g := s.group(key) - if _, duplicate := g.resources[plan.Key]; mutation && duplicate { + folds, mutated := g.resources[plan.Key] + if mutation && mutated && (!folds || !plan.Folds) { + if ordered { + s.ordered[resource] = nil + } continue } if len(g.items) >= r.limits.BatchOperations || len(g.items) > 0 && plan.Bytes > r.limits.BatchBytes-g.bytes { g.full = true + if ordered { + s.ordered[resource] = nil + } continue } g.items = append(g.items, t) g.bytes += plan.Bytes if mutation { if g.resources == nil { - g.resources = make(map[string]struct{}) + g.resources = make(map[string]bool) } - g.resources[plan.Key] = struct{}{} + g.resources[plan.Key] = plan.Folds + if mutated { + g.folded++ + } + } + if ordered { + s.ordered[resource] = g } } for _, g := range s.groups[:s.used] { @@ -479,6 +504,7 @@ func (r *Runtime) selectLocked(now time.Time) ([]*Ticket, time.Time) { // The batch owns the returned slice. items := g.items r.fullNext = r.isFull(g) + r.foldedNext = g.folded g.items = nil var ready time.Time return items, ready @@ -576,7 +602,10 @@ func (r *Runtime) launchLocked(items []*Ticket) *batch { } b := newBatch(items) b.probe, b.full = r.probeNext, r.fullNext - r.probeNext, r.fullNext = false, false + if r.foldedNext > 0 { + r.metrics.folded.Add(float64(r.foldedNext)) + } + r.probeNext, r.fullNext, r.foldedNext = false, false, 0 r.batches[b] = struct{}{} if b.native() { r.natives++ diff --git a/internal/store/runtime_test.go b/internal/store/runtime_test.go index a0871731..d7709053 100644 --- a/internal/store/runtime_test.go +++ b/internal/store/runtime_test.go @@ -13,6 +13,7 @@ import ( "github.com/batchstream/weir-protocol/api/protocol" pb "github.com/batchstream/weir-protocol/api/weir/v1" "github.com/batchstream/weir/internal/execution" + "github.com/batchstream/weir/internal/testutil/testmetrics" "google.golang.org/protobuf/proto" ) @@ -131,6 +132,13 @@ func mutationPlan(resource string) *execution.Plan { return plan } +// foldingPlan is a mutation the adapter can fold with others of its resource. +func foldingPlan(resource string) *execution.Plan { + plan := mutationPlan(resource) + plan.Folds = true + return plan +} + func nativeCommand() *pb.Command { request := &pb.NativeRequest{Resource: "db"} value := &pb.Command_Native{Native: request} @@ -427,6 +435,141 @@ func TestSameResourceMutationsNeverShareABatch(t *testing.T) { } } +// queueBehindBlocker holds the only execution slot with a running batch, so +// that the plans submitted next queue together; release lets them run. +func queueBehindBlocker(t *testing.T, change func(*Limits)) (*fakeAdapter, *Runtime, func()) { + t.Helper() + adapter := &fakeAdapter{} + release := make(chan struct{}) + runtime := open(t, adapter, limits(func(l *Limits) { + l.MaxInFlight = 1 + if change != nil { + change(l) + } + })) + adapter.records = blockUntil(release) + blocker := submit(t, runtime, mutationPlan("blocker"), nil) + time.Sleep(20 * time.Millisecond) + finish := func() { + close(release) + wait(t, blocker) + } + return adapter, runtime, finish +} + +func TestFoldingMutationsOfOneResourceShareABatchInQueueOrder(t *testing.T) { + adapter, runtime, finish := queueBehindBlocker(t, nil) + session, other := runtime.NewSession(), runtime.NewSession() + first, second, third := foldingPlan("x"), foldingPlan("x"), foldingPlan("x") + tickets := []*Ticket{ + submit(t, runtime, first, session), + submit(t, runtime, foldingPlan("y"), session), + submit(t, runtime, second, other), + submit(t, runtime, third, session), + } + finish() + for _, ticket := range tickets { + wait(t, ticket) + } + want := [][]string{{"blocker"}, {"x", "y", "x", "x"}} + if got := adapter.executed(); !slices.EqualFunc(got, want, slices.Equal) { + t.Fatal("folding mutations did not share one batch", got) + } + batch := adapter.batches[1] + if batch[0] != first || batch[2] != second || batch[3] != third { + t.Fatal("folding mutations lost their queue order") + } + families := testmetrics.Gather(t, runtime) + if got := counter(families, "weir_store_folded_mutations_total", nil); got != 2 { + t.Fatal("folded mutations counted", got) + } +} + +func TestAMutationThatDoesNotFoldEndsItsResourceChain(t *testing.T) { + adapter, runtime, finish := queueBehindBlocker(t, nil) + session := runtime.NewSession() + tickets := []*Ticket{ + submit(t, runtime, foldingPlan("x"), session), + submit(t, runtime, mutationPlan("x"), session), + submit(t, runtime, foldingPlan("x"), session), + submit(t, runtime, foldingPlan("y"), runtime.NewSession()), + } + finish() + for _, ticket := range tickets { + wait(t, ticket) + } + want := [][]string{{"blocker"}, {"x", "y"}, {"x"}, {"x"}} + if got := adapter.executed(); !slices.EqualFunc(got, want, slices.Equal) { + t.Fatal("a mutation that does not fold joined or was passed", got) + } + if adapter.batches[2][0].Folds { + t.Fatal("the folding mutation passed the one before it") + } +} + +func TestSessionChainsStayInTheirGroup(t *testing.T) { + adapter, runtime, finish := queueBehindBlocker(t, nil) + session := runtime.NewSession() + visible := foldingPlan("x") + visible.BatchKey = "visible" + tickets := []*Ticket{ + submit(t, runtime, foldingPlan("x"), session), + submit(t, runtime, visible, session), + submit(t, runtime, foldingPlan("x"), session), + } + finish() + for _, ticket := range tickets { + wait(t, ticket) + } + want := [][]string{{"blocker"}, {"x"}, {"x"}, {"x"}} + if got := adapter.executed(); !slices.EqualFunc(got, want, slices.Equal) { + t.Fatal("a session chain crossed batch keys", got) + } + if adapter.batches[2][0] != visible { + t.Fatal("session order changed across batch keys") + } +} + +func TestFullBatchesEndSessionChains(t *testing.T) { + adapter, runtime, finish := queueBehindBlocker(t, func(l *Limits) { l.BatchOperations = 2 }) + session := runtime.NewSession() + var tickets []*Ticket + for _, key := range []string{"a", "x", "x", "x"} { + tickets = append(tickets, submit(t, runtime, foldingPlan(key), session)) + } + finish() + for _, ticket := range tickets { + wait(t, ticket) + } + want := [][]string{{"blocker"}, {"a", "x"}, {"x", "x"}} + if got := adapter.executed(); !slices.EqualFunc(got, want, slices.Equal) { + t.Fatal("full batches reordered a session chain", got) + } +} + +func TestFoldingMutationsWaitForTheirRunningPredecessor(t *testing.T) { + adapter := &fakeAdapter{started: make(chan []*execution.Plan, 8)} + release := make(chan struct{}) + runtime := open(t, adapter, limits(nil)) + settle(runtime) + adapter.records = blockUntil(release) + session := runtime.NewSession() + first := submit(t, runtime, foldingPlan("x"), session) + if keys := started(t, adapter); !slices.Equal(keys, []string{"x"}) { + t.Fatal(keys) + } + second := submit(t, runtime, foldingPlan("x"), session) + third := submit(t, runtime, foldingPlan("x"), session) + notStarted(t, adapter) + close(release) + if keys := started(t, adapter); !slices.Equal(keys, []string{"x", "x"}) { + t.Fatal("queued folding mutations did not follow their predecessor together", keys) + } + for _, ticket := range []*Ticket{first, second, third} { + wait(t, ticket) + } +} + func TestCanceledQueuedMutationKeepsSessionOrder(t *testing.T) { adapter := &fakeAdapter{started: make(chan []*execution.Plan, 8)} release := make(chan struct{}) From 9dc85ea4888f2b4f5336c73ad7d7f307f4398179 Mon Sep 17 00:00:00 2001 From: Randy <857271370@qq.com> Date: Sun, 8 Nov 2026 10:42:44 +0800 Subject: [PATCH 2/3] =?UTF-8?q?=F0=9F=90=9B=20fix:=20settle=20folded=20Sea?= =?UTF-8?q?rch=20chains=20whose=20callers=20leave=20or=20whose=20write=20f?= =?UTF-8?q?ails?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Re-evaluate a chain without a mutation whose caller left before the write, fail mutations rejected on a state only a failed write would have made real, replay one at a time only when the combined document is rejected, keep folded mutations out of the adaptive limit's batch sizes, and update tests that assumed one native write per increment. --- docs/batching.md | 30 +-- docs/operations.md | 2 +- internal/app/diagnostics_integration_test.go | 4 +- internal/backend/search/batch.go | 27 ++- internal/backend/search/fold.go | 67 ++++--- internal/backend/search/fold_test.go | 179 +++++++++++++++++- .../batching_scenarios_integration_test.go | 9 +- internal/server/lua_batch_integration_test.go | 14 +- internal/store/metrics.go | 2 +- internal/store/runtime.go | 33 ++-- internal/store/runtime_test.go | 55 ++++++ 11 files changed, 349 insertions(+), 73 deletions(-) diff --git a/docs/batching.md b/docs/batching.md index 58d892f7..46cfeee9 100644 --- a/docs/batching.md +++ b/docs/batching.md @@ -178,17 +178,25 @@ 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. A batch holds one mutation per resource, -except that Search folds the mutations of one record, apart from backend -expressions: those queued from one or several streams share a batch in queue -order, and Search applies them in that order to the record's state and sends one -write. The write is conditional on a real-time read unless the first mutation is -a put or delete, and query-visible mutations wait for one refresh. Each mutation -keeps its own result. One whose precondition fails or whose Lua rejects leaves -the state to the next. A conflict evaluates the chain again from a fresh read. -Any other rejection of the combined write runs the mutations again one at a -time, since none applied; an unknown write leaves every mutation it carried -unknown. `weir_store_folded_mutations_total` counts mutations that followed an -earlier one of the same resource in their batch. Grouping is not an application transaction: +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. 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. +`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. diff --git a/docs/operations.md b/docs/operations.md index 3841ee49..7d104238 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -97,7 +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, which the adapter [folds](batching.md) into one write | +| `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 | diff --git a/internal/app/diagnostics_integration_test.go b/internal/app/diagnostics_integration_test.go index 73e63855..8115534c 100644 --- a/internal/app/diagnostics_integration_test.go +++ b/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) } diff --git a/internal/backend/search/batch.go b/internal/backend/search/batch.go index cfe00735..befaddb0 100644 --- a/internal/backend/search/batch.go +++ b/internal/backend/search/batch.go @@ -183,11 +183,14 @@ type recordBatch struct { // 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. unfold lists chains whose write was rejected: nothing of them - // applied, so their mutations run again one at a time. - chains map[int][]int - applied map[int][]int - unfold []int + // 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. @@ -202,6 +205,7 @@ func (b *recordBatch) appendWrite(work *execution.Plan, position int, current *g if err != nil { if b.applied[position] != nil { delete(b.applied, position) + delete(b.dependent, position) b.unfold = append(b.unfold, position) return } @@ -279,9 +283,14 @@ func (b *recordBatch) flush() { positions := make([]int, 0, len(b.pending)) for _, item := range b.pending { if members := b.applied[item.position]; members != nil { - // A chain's write goes out while any of its callers remains. - if b.endChain(members) { + 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 { @@ -502,8 +511,8 @@ func (b *recordBatch) clear(i int) { } // 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 -// replaces or deletes the record whatever it holds. +// 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 { diff --git a/internal/backend/search/fold.go b/internal/backend/search/fold.go index 5f39fbe1..f566a944 100644 --- a/internal/backend/search/fold.go +++ b/internal/backend/search/fold.go @@ -31,8 +31,8 @@ func (b *recordBatch) live(head int) bool { // mutateChain applies a chain's mutations in queue order to the record's // state and frames one write that takes the record from the state it read, or -// from any state when the first mutation replaces or deletes it regardless, -// to the last. A mutation that is rejected leaves the state to the next one. +// from any state when the first mutation is a put or delete, to the last. A +// mutation that is rejected leaves the state to the next one. func (b *recordBatch) mutateChain(head int, current *getReply) { if current == nil && b.preReads(head) { // The mutation that made the read unnecessary lost its caller. @@ -44,7 +44,7 @@ func (b *recordBatch) mutateChain(head int, current *getReply) { if current != nil { found, source = *current.Found, current.Source } - var applied []int + var applied, dependent []int for _, i := range b.chains[head] { work := b.works[i] if denied := work.Ended(b.ctx); denied != nil { @@ -75,7 +75,8 @@ func (b *recordBatch) mutateChain(head int, current *getReply) { state := &getReply{Found: &present, Source: source} next, mutation := b.adapter.evaluateProgram(work, state) if denied := work.Ended(b.ctx); denied != nil { - mutation = protocol.Mutation(pb.MutationOutcome_NOT_APPLIED, denied) + b.fail(i, pb.MutationOutcome_NOT_APPLIED, denied) + continue } switch { case mutation == nil && next.action == "delete": @@ -88,6 +89,9 @@ func (b *recordBatch) mutateChain(head int, current *getReply) { } if rejected != nil { b.fail(i, rejected.Outcome, rejected.Failure) + if len(applied) != 0 { + dependent = append(dependent, i) + } continue } applied = append(applied, i) @@ -126,48 +130,55 @@ func (b *recordBatch) mutateChain(head int, current *getReply) { framed.Backend = write if b.applied == nil { b.applied = make(map[int][]int) + b.dependent = make(map[int][]int) } b.applied[head] = applied + b.dependent[head] = dependent b.appendWrite(&framed, head, current) } -// endChain fails a framed chain whose callers all left and reports whether -// it did. -func (b *recordBatch) endChain(members []int) bool { +// departed reports whether a mutation whose effect a framed chain's write +// carries has lost its caller. +func (b *recordBatch) departed(members []int) bool { for _, i := range members { - if b.works[i].Ended(b.ctx) == nil { - return false + if b.works[i].Ended(b.ctx) != nil { + return true } } - for _, i := range members { - b.fail(i, pb.MutationOutcome_NOT_APPLIED, b.works[i].Ended(b.ctx)) - } - return true + return false } // settleChain gives a chain's write result to each mutation the write -// carried. A conflict evaluates the chain again from a fresh read, and any -// other rejection runs its mutations one at a time, since none applied. +// carried. A conflict evaluates the chain again from a fresh read, and a +// rejected document runs its mutations one at a time, since none applied. A +// write that did not apply also fails the mutations rejected on the state it +// would have made real. func (b *recordBatch) settleChain(head int, reply *pb.Event) { - members := b.applied[head] + members, dependent := b.applied[head], b.dependent[head] delete(b.applied, head) + delete(b.dependent, head) mutation := reply.GetMutationResult() - if mutation.GetOutcome() == pb.MutationOutcome_NOT_APPLIED { - if mutation.GetFailure().GetCode() != pb.FailureCode_CONFLICT { + outcome, failure := mutation.GetOutcome(), mutation.GetFailure() + if outcome == pb.MutationOutcome_NOT_APPLIED { + switch failure.GetCode() { + case pb.FailureCode_PRECONDITION_FAILED: b.unfold = append(b.unfold, head) return + case pb.FailureCode_CONFLICT: + if b.attempts[head] < b.adapter.options().RetryAttempts { + b.retry = append(b.retry, head) + return + } + failure = protocol.Fail(pb.FailureCode_CONFLICT, "Search record changed during every transform attempt") } - if b.attempts[head] < b.adapter.options().RetryAttempts { - b.retry = append(b.retry, head) - return - } - failure := protocol.Fail(pb.FailureCode_CONFLICT, "Search record changed during every transform attempt") - for _, i := range members { - b.fail(i, pb.MutationOutcome_NOT_APPLIED, failure) - } - return } for _, i := range members { - b.fail(i, mutation.GetOutcome(), mutation.GetFailure()) + b.fail(i, outcome, failure) + } + if outcome == pb.MutationOutcome_APPLIED { + return + } + for _, i := range dependent { + b.fail(i, pb.MutationOutcome_NOT_APPLIED, failure) } } diff --git a/internal/backend/search/fold_test.go b/internal/backend/search/fold_test.go index df30246a..b3a7566f 100644 --- a/internal/backend/search/fold_test.go +++ b/internal/backend/search/fold_test.go @@ -13,6 +13,7 @@ import ( "testing" pb "github.com/batchstream/weir-protocol/api/weir/v1" + "github.com/batchstream/weir/internal/backend" "github.com/batchstream/weir/internal/execution" ) @@ -32,9 +33,15 @@ type memorySearch struct { refreshes []string barriers int // changeBeforeWrite rewrites a stored record before the next bulk applies, - // as a concurrent writer would. + // as a concurrent writer would, or before every bulk with keepChanging. changeBeforeWrite []string + keepChanging bool reject func(source string) bool + // onRead runs with the IDs of each real-time read, and bulkStatus, when + // set, answers every bulk with that status and bulkBody. + onRead func(ids []string) + bulkStatus int + bulkBody string } func (m *memorySearch) handle(w http.ResponseWriter, r *http.Request) { @@ -47,6 +54,9 @@ func (m *memorySearch) handle(w http.ResponseWriter, r *http.Request) { m.reads++ var request struct{ IDs []string } _ = json.NewDecoder(r.Body).Decode(&request) + if m.onRead != nil { + m.onRead(request.IDs) + } docs := make([]string, len(request.IDs)) for i, id := range request.IDs { doc, found := m.docs[id] @@ -64,11 +74,18 @@ func (m *memorySearch) handle(w http.ResponseWriter, r *http.Request) { raw, _ := io.ReadAll(r.Body) m.bulks = append(m.bulks, string(raw)) m.refreshes = append(m.refreshes, r.URL.Query().Get("refresh")) + if m.bulkStatus != 0 { + w.WriteHeader(m.bulkStatus) + _, _ = io.WriteString(w, m.bulkBody) + return + } for _, id := range m.changeBeforeWrite { m.seq++ m.docs[id] = storedSource{source: `{"n":100}`, seq: m.seq} } - m.changeBeforeWrite = nil + if !m.keepChanging { + m.changeBeforeWrite = nil + } items, failed := m.apply(string(raw)) _, _ = fmt.Fprintf(w, `{"errors":%t,"took":1,"items":[%s]}`, failed, strings.Join(items, ",")) default: @@ -354,11 +371,165 @@ func TestBackendExpressionsDoNotFold(t *testing.T) { t.Fatal("a backend expression cannot be applied to a known source") } for _, action := range []string{"put", "create", "replace", "delete", "program"} { - if work := batchTestPlan(t, adapter, action, "records/s:x"); !work.Folds { + work := batchTestPlan(t, adapter, action, "records/s:x") + if !work.Folds { t.Fatal(action, "does not fold") } } - if read := batchTestPlan(t, adapter, "read", "records/s:x"); read.Folds || !slices.Contains([]string{"records"}, read.BatchKey) { + read := batchTestPlan(t, adapter, "read", "records/s:x") + if read.Folds || read.BatchKey != "records" { t.Fatal("a read folds") } } + +// cancelOnRead returns a caller context that ends when a read of id begins. +func cancelOnRead(t *testing.T, memory *memorySearch, id string) context.Context { + caller, cancel := context.WithCancel(t.Context()) + t.Cleanup(cancel) + memory.onRead = func(ids []string) { + if slices.Contains(ids, id) { + cancel() + } + } + return caller +} + +// onePreReadPerRound makes each round of a batch read at most one record, so +// a later round can run between a chain's evaluation and its write. +func onePreReadPerRound(adapter *Adapter) { + settings := backend.DefaultOptions() + settings.BatchBytes = settings.ExchangeBytes + adapter.config.Options = &settings +} + +func TestAChainWhoseMemberLeftBeforeItsWriteRunsAgainWithoutIt(t *testing.T) { + memory, adapter := newMemorySearch(t, map[string]string{"y": `{"n":1}`}) + onePreReadPerRound(adapter) + x := recordMutations{t: t, adapter: adapter, resource: "records/s:x"} + y := recordMutations{t: t, adapter: adapter, resource: "records/s:y"} + works := []*execution.Plan{ + x.prepare("add", `{"by":1}`), + x.prepare("create", `{"n":50}`), + y.prepare("add", `{"by":1}`), + } + // The first caller leaves once its chain is evaluated, before the write. + works[0].Context = cancelOnRead(t, memory, "y") + results := adapter.ExecuteRecords(t.Context(), works) + // The create was rejected on the state the departed add left; without it, + // the create applies. + canceled := expected{outcome: pb.MutationOutcome_NOT_APPLIED, code: pb.FailureCode_CANCELLED} + want := []expected{canceled, appliedResult, appliedResult} + checkResults(t, results, want) + if got := memory.source("x"); got != `{"n":50}` { + t.Fatal("the departed mutation's effect was written", got) + } +} + +func TestAChainWhoseReaderLeftReadsAgainForTheOthers(t *testing.T) { + memory, adapter := newMemorySearch(t, map[string]string{"x": `{"n":1}`}) + x := recordMutations{t: t, adapter: adapter, resource: "records/s:x"} + works := []*execution.Plan{ + x.prepare("add", `{"by":1}`), + x.prepare("add", `{"by":10}`), + } + works[0].Context = cancelOnRead(t, memory, "x") + results := adapter.ExecuteRecords(t.Context(), works) + canceled := expected{outcome: pb.MutationOutcome_NOT_APPLIED, code: pb.FailureCode_CANCELLED} + want := []expected{canceled, appliedResult} + checkResults(t, results, want) + if got := memory.source("x"); got != `{"n":11}` || memory.reads != 2 { + t.Fatal("source", got, "reads", memory.reads) + } +} + +func TestAChainWhoseLeadingPutLeftReadsTheRecord(t *testing.T) { + memory, adapter := newMemorySearch(t, map[string]string{"x": `{"n":1}`, "y": `{"n":1}`}) + x := recordMutations{t: t, adapter: adapter, resource: "records/s:x"} + y := recordMutations{t: t, adapter: adapter, resource: "records/s:y"} + works := []*execution.Plan{ + x.prepare("put", `{"n":5}`), + x.prepare("add", `{"by":10}`), + y.prepare("add", `{"by":1}`), + } + // The put made x's read unnecessary until its caller left. + works[0].Context = cancelOnRead(t, memory, "y") + results := adapter.ExecuteRecords(t.Context(), works) + canceled := expected{outcome: pb.MutationOutcome_NOT_APPLIED, code: pb.FailureCode_CANCELLED} + want := []expected{canceled, appliedResult, appliedResult} + checkResults(t, results, want) + if got := memory.source("x"); got != `{"n":11}` { + t.Fatal("the transform did not apply to the stored record", got) + } +} + +func TestAFoldedWriteThatKeepsConflictingFailsEveryMutationItStoodFor(t *testing.T) { + memory, adapter := newMemorySearch(t, nil) + memory.changeBeforeWrite, memory.keepChanging = []string{"x"}, true + settings := backend.DefaultOptions() + settings.RetryAttempts = 2 + adapter.config.Options = &settings + x := recordMutations{t: t, adapter: adapter, resource: "records/s:x"} + works := []*execution.Plan{ + x.prepare("add", `{"by":1}`), + x.prepare("create", `{"n":9}`), + } + results := adapter.ExecuteRecords(t.Context(), works) + // The create was rejected on a state no write made real. + conflict := expected{outcome: pb.MutationOutcome_NOT_APPLIED, code: pb.FailureCode_CONFLICT} + want := []expected{conflict, conflict} + checkResults(t, results, want) + if len(memory.bulks) != 2 { + t.Fatal("bulks", len(memory.bulks)) + } +} + +func TestAnUnknownFoldedWriteLeavesItsDependentsNotApplied(t *testing.T) { + memory, adapter := newMemorySearch(t, nil) + memory.bulkStatus, memory.bulkBody = http.StatusInternalServerError, `{}` + x := recordMutations{t: t, adapter: adapter, resource: "records/s:x"} + works := []*execution.Plan{ + x.prepare("add", `{"by":1}`), + x.prepare("create", `{"n":9}`), + } + results := adapter.ExecuteRecords(t.Context(), works) + unknown := expected{outcome: pb.MutationOutcome_UNKNOWN, code: pb.FailureCode_UNAVAILABLE} + notApplied := expected{outcome: pb.MutationOutcome_NOT_APPLIED, code: pb.FailureCode_UNAVAILABLE} + want := []expected{unknown, notApplied} + checkResults(t, results, want) +} + +func TestACapacityRejectionOfAFoldedWriteDoesNotRunItsMutationsAlone(t *testing.T) { + memory, adapter := newMemorySearch(t, map[string]string{"x": `{"n":1}`, "y": `{"n":1}`}) + memory.bulkStatus = http.StatusTooManyRequests + memory.bulkBody = `{"error":{"type":"es_rejected_execution_exception"},"status":429}` + x := recordMutations{t: t, adapter: adapter, resource: "records/s:x"} + y := recordMutations{t: t, adapter: adapter, resource: "records/s:y"} + works := []*execution.Plan{ + x.prepare("add", `{"by":1}`), + y.prepare("add", `{"by":1}`), + x.prepare("add", `{"by":1}`), + y.prepare("add", `{"by":1}`), + } + results := adapter.ExecuteRecords(t.Context(), works) + unavailable := expected{outcome: pb.MutationOutcome_NOT_APPLIED, code: pb.FailureCode_UNAVAILABLE} + want := []expected{unavailable, unavailable, unavailable, unavailable} + checkResults(t, results, want) + if memory.reads != 1 || len(memory.bulks) != 1 { + t.Fatal("reads", memory.reads, "bulks", len(memory.bulks)) + } +} + +func TestAFoldedDeleteOfAMissingRecordWaitsForARefresh(t *testing.T) { + memory, adapter := newMemorySearch(t, nil) + x := recordMutations{t: t, adapter: adapter, resource: "records/s:x", visible: true} + works := []*execution.Plan{ + x.prepare("put", `{"n":1}`), + x.prepare("delete", ``), + } + results := adapter.ExecuteRecords(t.Context(), works) + want := []expected{appliedResult, appliedResult} + checkResults(t, results, want) + if len(memory.bulks) != 1 || !strings.Contains(memory.bulks[0], `"delete"`) || memory.barriers != 1 { + t.Fatal("bulks", memory.bulks, "refresh barriers", memory.barriers) + } +} diff --git a/internal/server/batching_scenarios_integration_test.go b/internal/server/batching_scenarios_integration_test.go index 3c5cc55f..6278a0aa 100644 --- a/internal/server/batching_scenarios_integration_test.go +++ b/internal/server/batching_scenarios_integration_test.go @@ -130,6 +130,10 @@ func incrementScenarioRecords(t *testing.T, node *routeAcceptanceNode, records [ return results } +// verifyScenarioRecord checks a record's counter and untouched fields after +// increments applied ones. Search folds increments of one record that share a +// batch into one write, so each write raises the version once for one or more +// of them. func verifyScenarioRecord(t *testing.T, fixture *testsearch.Backend, record scenarioRecord, increments int) { t.Helper() code, raw := fixture.Do(t, http.MethodGet, "/"+record.index+"/_doc/"+record.id, "") @@ -142,7 +146,8 @@ func verifyScenarioRecord(t *testing.T, fixture *testsearch.Backend, record scen } wantPadding, _ := json.Marshal(strings.Repeat("x", record.padding)) wantMarker, _ := json.Marshal(record.id) - if reply.Version != increments+1 || string(reply.Source["n"]) != fmt.Sprint(increments) || !bytes.Equal(reply.Source["pad"], wantPadding) || !bytes.Equal(reply.Source["marker"], wantMarker) { + writes := reply.Version - 1 + if writes > increments || writes < min(increments, 1) || string(reply.Source["n"]) != fmt.Sprint(increments) || !bytes.Equal(reply.Source["pad"], wantPadding) || !bytes.Equal(reply.Source["marker"], wantMarker) { t.Fatal("persisted version, counter or untouched fields differ", record.id, reply.Version, increments) } } @@ -235,5 +240,5 @@ func TestRouteSearchBoundedCollectionPreservesHotKeyIncrements(t *testing.T) { t.Fatal("no shared-key write applied") } verifyScenarioRecord(t, fixture, hot, applied) - t.Logf("10%% shared-key increments, 4 execution slots and 20ms collection: %d/8 shared mutations applied; persisted counter and version match exactly", applied) + t.Logf("10%% shared-key increments, 4 execution slots and 20ms collection: %d/8 shared mutations applied; persisted counter matches exactly and the version counts at most one write per increment", applied) } diff --git a/internal/server/lua_batch_integration_test.go b/internal/server/lua_batch_integration_test.go index cfdc156a..fc0b4fc4 100644 --- a/internal/server/lua_batch_integration_test.go +++ b/internal/server/lua_batch_integration_test.go @@ -473,11 +473,17 @@ func TestRouteSearchLuaSameURIRPCsPreserveAppliedIncrements(t *testing.T) { Version int64 `json:"_version"` Source map[string]json.RawMessage `json:"_source"` } - if code != http.StatusOK || json.Unmarshal(raw, &stored) != nil || len(stored.Source) != 2 || stored.Version != int64(applied+1) || string(stored.Source["n"]) != fmt.Sprint(applied) || string(stored.Source["marker"]) != `"untouched"` { + if code != http.StatusOK || json.Unmarshal(raw, &stored) != nil || len(stored.Source) != 2 || string(stored.Source["n"]) != fmt.Sprint(applied) || string(stored.Source["marker"]) != `"untouched"` { t.Fatal("concurrent same-URI Lua RPCs lost updates or added metadata", code, string(raw)) } - if observer.reads.Load() < int32(len(cases)+1) || observer.writes.Load() < int32(applied) { - t.Fatal("same-URI Lua RPCs did not use successive OCC snapshots", observer.reads.Load(), observer.writes.Load()) + // Increments that share a batch fold into one write, so the version + // counts writes, at most one per applied increment. + writes := stored.Version - 1 + if applied == 0 || writes < 1 || writes > int64(applied) { + t.Fatal("same-URI Lua RPC writes do not match APPLIED outcomes", applied, string(raw)) } - t.Log("16 concurrent same-URI Lua RPCs: native versions and persisted increments match APPLIED outcomes; source fields unchanged") + if observer.reads.Load() < 2 || observer.writes.Load() < int32(writes) { + t.Fatal("same-URI Lua RPCs did not write from OCC snapshots", observer.reads.Load(), observer.writes.Load()) + } + t.Log("16 concurrent same-URI Lua RPCs: persisted increments match APPLIED outcomes, with at most one native write each; source fields unchanged") } diff --git a/internal/store/metrics.go b/internal/store/metrics.go index 0e3d9afc..2e69676a 100644 --- a/internal/store/metrics.go +++ b/internal/store/metrics.go @@ -19,7 +19,7 @@ func newRuntimeMetrics() runtimeMetrics { queueOptions := prometheus.HistogramOpts{Name: "weir_store_queue_wait_seconds", Help: "Record time from queueing to dispatch.", Buckets: []float64{.001, .01, .1, 1, 10}} durationOptions := prometheus.HistogramOpts{Name: "weir_store_execution_seconds", Help: "Record batch execution time.", Buckets: []float64{.001, .01, .1, 1, 10}} nativeDurationOptions := prometheus.HistogramOpts{Name: "weir_store_native_seconds", Help: "Native call time, including sending the response on the RPC stream.", Buckets: []float64{.001, .01, .1, 1, 10}} - foldedOptions := prometheus.CounterOpts{Name: "weir_store_folded_mutations_total", Help: "Mutations dispatched in a batch after an earlier mutation of the same resource, which the adapter applies with it."} + foldedOptions := prometheus.CounterOpts{Name: "weir_store_folded_mutations_total", Help: "Mutations dispatched in a batch after an earlier mutation of the same resource, for the adapter to fold into one write."} batchOptions := prometheus.HistogramOpts{Name: "weir_store_batch_operations", Help: "Records per record batch.", Buckets: prometheus.ExponentialBuckets(1, 2, 11)} metrics := runtimeMetrics{ records: prometheus.NewCounterVec(recordOptions, []string{"operation", "outcome"}), diff --git a/internal/store/runtime.go b/internal/store/runtime.go index b1decd80..84fe0574 100644 --- a/internal/store/runtime.go +++ b/internal/store/runtime.go @@ -69,10 +69,9 @@ type Runtime struct { limit [2]executionLimit probes probes // probeNext and fullNext describe the batch about to launch: the limit's - // probe, or one that is full. foldedNext counts its mutations that follow - // an earlier one of the same resource. + // probe, or one that is full. foldedNext counts its folded mutations. probeNext, fullNext bool - foldedNext int + foldedNext folds draining, closed, overloaded bool // space and nativeSpace are closed and replaced whenever a waiting record // or Native submitter may proceed. @@ -127,6 +126,14 @@ type batch struct { // probe marks the limit's probe, full a batch of max_operations records or // max_bytes, and partial one the adaptive limit counts. probe, full, partial bool + folded folds +} + +// folds counts the mutations of a batch that follow an earlier one of the +// same resource, and their bytes. The adapter folds them into that earlier +// one's write, so they do not make the batch's execution longer. +type folds struct { + count, bytes int } func New(adapter execution.Adapter, limits Limits) (*Runtime, error) { @@ -388,7 +395,7 @@ type group struct { // far all fold. Mutations of one resource share a batch only when they // all fold. resources map[string]bool - folded int + folded folds } type selection struct { @@ -407,7 +414,7 @@ func (s *selection) reset() { for _, g := range s.groups[:s.used] { clear(g.items) g.items = g.items[:0] - g.bytes, g.full, g.folded = 0, false, 0 + g.bytes, g.full, g.folded = 0, false, folds{} clear(g.resources) } s.used = 0 @@ -482,7 +489,8 @@ func (r *Runtime) selectLocked(now time.Time) ([]*Ticket, time.Time) { } g.resources[plan.Key] = plan.Folds if mutated { - g.folded++ + g.folded.count++ + g.folded.bytes += plan.Bytes } } if ordered { @@ -601,11 +609,11 @@ func (r *Runtime) launchLocked(items []*Ticket) *batch { t.stop() } b := newBatch(items) - b.probe, b.full = r.probeNext, r.fullNext - if r.foldedNext > 0 { - r.metrics.folded.Add(float64(r.foldedNext)) + b.probe, b.full, b.folded = r.probeNext, r.fullNext, r.foldedNext + if b.folded.count > 0 { + r.metrics.folded.Add(float64(b.folded.count)) } - r.probeNext, r.fullNext, r.foldedNext = false, false, 0 + r.probeNext, r.fullNext, r.foldedNext = false, false, folds{} r.batches[b] = struct{}{} if b.native() { r.natives++ @@ -727,7 +735,10 @@ func (r *Runtime) execute(b *batch) { for _, t := range b.items { bytes += t.plan.Bytes } - done := finishedBatch{started: started, finished: finished, key: key, size: len(b.items), bytes: bytes, full: b.full, probe: b.probe} + // Folded mutations add no backend work of their own, so a batch + // counts as the size it would be without them. + size, bytes := len(b.items)-b.folded.count, bytes-b.folded.bytes + done := finishedBatch{started: started, finished: finished, key: key, size: size, bytes: bytes, full: b.full, probe: b.probe} r.limit[op].observe(done) } if r.adaptive && r.running[op] == 0 && !r.queuesLocked(op) { diff --git a/internal/store/runtime_test.go b/internal/store/runtime_test.go index d7709053..122249f1 100644 --- a/internal/store/runtime_test.go +++ b/internal/store/runtime_test.go @@ -73,6 +73,12 @@ func (a *fakeAdapter) Close(ctx context.Context) error { return a.closeErr } +func (a *fakeAdapter) batch(i int) []*execution.Plan { + a.mu.Lock() + defer a.mu.Unlock() + return a.batches[i] +} + func (a *fakeAdapter) executed() [][]string { a.mu.Lock() defer a.mu.Unlock() @@ -570,6 +576,55 @@ func TestFoldingMutationsWaitForTheirRunningPredecessor(t *testing.T) { } } +func TestMutationsWithoutASessionFoldWhenTheyAllFold(t *testing.T) { + adapter, runtime, finish := queueBehindBlocker(t, nil) + tickets := []*Ticket{ + submit(t, runtime, foldingPlan("x"), nil), + submit(t, runtime, mutationPlan("y"), nil), + submit(t, runtime, foldingPlan("x"), nil), + submit(t, runtime, mutationPlan("y"), nil), + } + finish() + for _, ticket := range tickets { + wait(t, ticket) + } + want := [][]string{{"blocker"}, {"x", "y", "x"}, {"y"}} + if got := adapter.executed(); !slices.EqualFunc(got, want, slices.Equal) { + t.Fatal("mutations without a session folded wrongly", got) + } + families := testmetrics.Gather(t, runtime) + if got := counter(families, "weir_store_folded_mutations_total", nil); got != 1 { + t.Fatal("only the folding duplicate counts", got) + } +} + +func TestAnotherSessionsMutationDoesNotWaitOrFoldWithARunningOne(t *testing.T) { + adapter := &fakeAdapter{started: make(chan []*execution.Plan, 8)} + release := make(chan struct{}) + runtime := open(t, adapter, limits(nil)) + settle(runtime) + adapter.records = blockUntil(release) + session, other := runtime.NewSession(), runtime.NewSession() + first := submit(t, runtime, foldingPlan("x"), session) + if keys := started(t, adapter); !slices.Equal(keys, []string{"x"}) { + t.Fatal(keys) + } + second := submit(t, runtime, foldingPlan("x"), session) + // The other session's mutation does not wait for the first session's. + third := submit(t, runtime, foldingPlan("x"), other) + if keys := started(t, adapter); !slices.Equal(keys, []string{"x"}) || adapter.batch(1)[0] != third.plan { + t.Fatal("the other session's mutation waited or took the queued one along", keys) + } + notStarted(t, adapter) + close(release) + if keys := started(t, adapter); !slices.Equal(keys, []string{"x"}) || adapter.batch(2)[0] != second.plan { + t.Fatal("the session's queued mutation did not follow its predecessor", keys) + } + for _, ticket := range []*Ticket{first, second, third} { + wait(t, ticket) + } +} + func TestCanceledQueuedMutationKeepsSessionOrder(t *testing.T) { adapter := &fakeAdapter{started: make(chan []*execution.Plan, 8)} release := make(chan struct{}) From b62607353e09084268dc7bd2d14c3154ca74935b Mon Sep 17 00:00:00 2001 From: Randy <857271370@qq.com> Date: Sun, 8 Nov 2026 11:13:44 +0800 Subject: [PATCH 3/3] =?UTF-8?q?=F0=9F=90=9B=20fix:=20keep=20no-ops=20and?= =?UTF-8?q?=20left-out=20mutations=20correct=20in=20folded=20batches?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A mutation decided on the record as read that leaves it unchanged keeps APPLIED whatever the chain's write returns, later mutations of a resource no longer join a batch that left out an earlier one, and the docs state that the backend validates only the chain's final document. --- docs/batching.md | 12 ++- internal/backend/search/fold.go | 15 +++- internal/backend/search/fold_test.go | 120 +++++++++++++++++++++++---- internal/store/runtime.go | 16 ++-- internal/store/runtime_test.go | 44 ++++++++++ 5 files changed, 182 insertions(+), 25 deletions(-) diff --git a/docs/batching.md b/docs/batching.md index 46cfeee9..7eac33e3 100644 --- a/docs/batching.md +++ b/docs/batching.md @@ -180,13 +180,17 @@ 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. 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 +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 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. diff --git a/internal/backend/search/fold.go b/internal/backend/search/fold.go index f566a944..a007e64e 100644 --- a/internal/backend/search/fold.go +++ b/internal/backend/search/fold.go @@ -44,6 +44,9 @@ func (b *recordBatch) mutateChain(head int, current *getReply) { if current != nil { found, source = *current.Found, current.Source } + // changed reports whether the state may differ from the record as read; + // without a read it is not known. + changed := current == nil var applied, dependent []int for _, i := range b.chains[head] { work := b.works[i] @@ -89,11 +92,21 @@ func (b *recordBatch) mutateChain(head int, current *getReply) { } if rejected != nil { b.fail(i, rejected.Outcome, rejected.Failure) - if len(applied) != 0 { + if changed { dependent = append(dependent, i) } continue } + if !changed && found == *current.Found && (!found || bytes.Equal(source, current.Source)) { + // Decided on the record as read and left it unchanged, so the + // result does not depend on the chain's write. + b.fail(i, pb.MutationOutcome_APPLIED, nil) + if step.waitUntilVisible { + b.queueVisibilityBarrier(step.index, i) + } + continue + } + changed = true applied = append(applied, i) } if len(applied) == 0 { diff --git a/internal/backend/search/fold_test.go b/internal/backend/search/fold_test.go index b3a7566f..0de6c688 100644 --- a/internal/backend/search/fold_test.go +++ b/internal/backend/search/fold_test.go @@ -33,15 +33,18 @@ type memorySearch struct { refreshes []string barriers int // changeBeforeWrite rewrites a stored record before the next bulk applies, - // as a concurrent writer would, or before every bulk with keepChanging. + // as a concurrent writer would, or before every bulk with keepChanging, + // once skipBulks bulks have passed. changeBeforeWrite []string keepChanging bool + skipBulks int reject func(source string) bool - // onRead runs with the IDs of each real-time read, and bulkStatus, when - // set, answers every bulk with that status and bulkBody. - onRead func(ids []string) - bulkStatus int - bulkBody string + // onRead runs with the IDs of each real-time read, and bulkStatus and + // refreshStatus, when set, answer every bulk or refresh with that status. + onRead func(ids []string) + bulkStatus int + bulkBody string + refreshStatus int } func (m *memorySearch) handle(w http.ResponseWriter, r *http.Request) { @@ -69,6 +72,11 @@ func (m *memorySearch) handle(w http.ResponseWriter, r *http.Request) { _, _ = io.WriteString(w, `{"docs":[`+strings.Join(docs, ",")+`]}`) case "/records/_refresh": m.barriers++ + if m.refreshStatus != 0 { + w.WriteHeader(m.refreshStatus) + _, _ = io.WriteString(w, `{}`) + return + } _, _ = io.WriteString(w, `{"_shards":{"total":1,"successful":1,"failed":0}}`) case "/_bulk": raw, _ := io.ReadAll(r.Body) @@ -79,12 +87,16 @@ func (m *memorySearch) handle(w http.ResponseWriter, r *http.Request) { _, _ = io.WriteString(w, m.bulkBody) return } - for _, id := range m.changeBeforeWrite { - m.seq++ - m.docs[id] = storedSource{source: `{"n":100}`, seq: m.seq} - } - if !m.keepChanging { - m.changeBeforeWrite = nil + if m.skipBulks > 0 { + m.skipBulks-- + } else { + for _, id := range m.changeBeforeWrite { + m.seq++ + m.docs[id] = storedSource{source: `{"n":100}`, seq: m.seq} + } + if !m.keepChanging { + m.changeBeforeWrite = nil + } } items, failed := m.apply(string(raw)) _, _ = fmt.Fprintf(w, `{"errors":%t,"took":1,"items":[%s]}`, failed, strings.Join(items, ",")) @@ -395,16 +407,18 @@ func cancelOnRead(t *testing.T, memory *memorySearch, id string) context.Context } // onePreReadPerRound makes each round of a batch read at most one record, so -// a later round can run between a chain's evaluation and its write. -func onePreReadPerRound(adapter *Adapter) { +// a later round can run between a chain's evaluation and its write, and +// allows attempts transforms of a record. +func onePreReadPerRound(adapter *Adapter, attempts int) { settings := backend.DefaultOptions() settings.BatchBytes = settings.ExchangeBytes + settings.RetryAttempts = attempts adapter.config.Options = &settings } func TestAChainWhoseMemberLeftBeforeItsWriteRunsAgainWithoutIt(t *testing.T) { memory, adapter := newMemorySearch(t, map[string]string{"y": `{"n":1}`}) - onePreReadPerRound(adapter) + onePreReadPerRound(adapter, backend.DefaultOptions().RetryAttempts) x := recordMutations{t: t, adapter: adapter, resource: "records/s:x"} y := recordMutations{t: t, adapter: adapter, resource: "records/s:y"} works := []*execution.Plan{ @@ -533,3 +547,79 @@ func TestAFoldedDeleteOfAMissingRecordWaitsForARefresh(t *testing.T) { t.Fatal("bulks", memory.bulks, "refresh barriers", memory.barriers) } } + +func TestALeavingCallerDoesNotUseUpAConflictAttempt(t *testing.T) { + memory, adapter := newMemorySearch(t, map[string]string{"x": `{"n":1}`, "y": `{"n":1}`}) + // The write of x, after y's, meets a concurrent change. + memory.changeBeforeWrite, memory.skipBulks = []string{"x"}, 1 + onePreReadPerRound(adapter, 2) + x := recordMutations{t: t, adapter: adapter, resource: "records/s:x"} + y := recordMutations{t: t, adapter: adapter, resource: "records/s:y"} + works := []*execution.Plan{ + x.prepare("add", `{"by":1}`), + x.prepare("add", `{"by":10}`), + y.prepare("add", `{"by":1}`), + } + works[0].Context = cancelOnRead(t, memory, "y") + results := adapter.ExecuteRecords(t.Context(), works) + canceled := expected{outcome: pb.MutationOutcome_NOT_APPLIED, code: pb.FailureCode_CANCELLED} + want := []expected{canceled, appliedResult, appliedResult} + checkResults(t, results, want) + if got := memory.source("x"); got != `{"n":110}` || len(memory.bulks) != 3 { + t.Fatal("source", got, "bulks", len(memory.bulks)) + } +} + +func TestOneChainConflictsWhileAnotherApplies(t *testing.T) { + memory, adapter := newMemorySearch(t, map[string]string{"x": `{"n":1}`, "y": `{"n":1}`}) + memory.changeBeforeWrite = []string{"x"} + x := recordMutations{t: t, adapter: adapter, resource: "records/s:x"} + y := recordMutations{t: t, adapter: adapter, resource: "records/s:y"} + works := []*execution.Plan{ + x.prepare("add", `{"by":1}`), + y.prepare("add", `{"by":1}`), + x.prepare("add", `{"by":1}`), + y.prepare("add", `{"by":1}`), + } + results := adapter.ExecuteRecords(t.Context(), works) + want := []expected{appliedResult, appliedResult, appliedResult, appliedResult} + checkResults(t, results, want) + if memory.source("x") != `{"n":102}` || memory.source("y") != `{"n":3}` || len(memory.bulks) != 2 { + t.Fatal("x", memory.source("x"), "y", memory.source("y"), "bulks", len(memory.bulks)) + } +} + +func TestANoopOnTheRecordAsReadKeepsItsResultWhateverTheWrite(t *testing.T) { + memory, adapter := newMemorySearch(t, map[string]string{"x": `{"n":1}`}) + memory.bulkStatus, memory.bulkBody = http.StatusInternalServerError, `{}` + x := recordMutations{t: t, adapter: adapter, resource: "records/s:x"} + works := []*execution.Plan{ + x.prepare("keep", `{}`), + x.prepare("create", `{"n":3}`), + x.prepare("replace", `{"n":2}`), + } + results := adapter.ExecuteRecords(t.Context(), works) + // The keep and the create were decided on the record as read, whatever + // the write did. + exists := expected{outcome: pb.MutationOutcome_NOT_APPLIED, code: pb.FailureCode_PRECONDITION_FAILED} + unknown := expected{outcome: pb.MutationOutcome_UNKNOWN, code: pb.FailureCode_UNAVAILABLE} + want := []expected{appliedResult, exists, unknown} + checkResults(t, results, want) +} + +func TestAFailedRefreshAfterAFoldedWriteFailsEveryMutationItCarried(t *testing.T) { + memory, adapter := newMemorySearch(t, nil) + memory.refreshStatus = http.StatusInternalServerError + x := recordMutations{t: t, adapter: adapter, resource: "records/s:x", visible: true} + works := []*execution.Plan{ + x.prepare("put", `{"n":1}`), + x.prepare("delete", ``), + } + results := adapter.ExecuteRecords(t.Context(), works) + for i, result := range results { + mutation := result.GetMutationResult() + if mutation.GetOutcome() != pb.MutationOutcome_APPLIED || mutation.GetFailure() == nil { + t.Fatal("mutation", i, mutation) + } + } +} diff --git a/internal/store/runtime.go b/internal/store/runtime.go index 84fe0574..3e415cac 100644 --- a/internal/store/runtime.go +++ b/internal/store/runtime.go @@ -391,9 +391,9 @@ type group struct { items []*Ticket bytes int full bool - // resources records each mutated resource and whether its mutations so - // far all fold. Mutations of one resource share a batch only when they - // all fold. + // resources records each mutated resource and whether a later mutation of + // it may still join: only while its mutations so far all fold and none + // was left out. resources map[string]bool folded folds } @@ -467,8 +467,11 @@ func (r *Runtime) selectLocked(now time.Time) ([]*Ticket, time.Time) { } } g := s.group(key) - folds, mutated := g.resources[plan.Key] - if mutation && mutated && (!folds || !plan.Folds) { + foldable, mutated := g.resources[plan.Key] + if mutation && mutated && (!foldable || !plan.Folds) { + // Later mutations of the resource, from any session, do not + // pass this one. + g.resources[plan.Key] = false if ordered { s.ordered[resource] = nil } @@ -476,6 +479,9 @@ func (r *Runtime) selectLocked(now time.Time) ([]*Ticket, time.Time) { } if len(g.items) >= r.limits.BatchOperations || len(g.items) > 0 && plan.Bytes > r.limits.BatchBytes-g.bytes { g.full = true + if mutation && mutated { + g.resources[plan.Key] = false + } if ordered { s.ordered[resource] = nil } diff --git a/internal/store/runtime_test.go b/internal/store/runtime_test.go index 122249f1..1b83e609 100644 --- a/internal/store/runtime_test.go +++ b/internal/store/runtime_test.go @@ -625,6 +625,50 @@ func TestAnotherSessionsMutationDoesNotWaitOrFoldWithARunningOne(t *testing.T) { } } +func TestAFoldedBatchTeachesTheLimitAsItsUnfoldedSize(t *testing.T) { + adapter := &fakeAdapter{} + runtime := open(t, adapter, limits(func(l *Limits) { l.BatchWait = 50 * time.Millisecond })) + var tickets []*Ticket + for range 8 { + tickets = append(tickets, submit(t, runtime, foldingPlan("x"), nil)) + } + for _, ticket := range tickets { + wait(t, ticket) + } + if got := adapter.executed(); len(got) != 1 { + t.Fatal("the folding mutations did not share one batch", got) + } + runtime.mu.Lock() + defer runtime.mu.Unlock() + key := groupKey{batch: "records"} + generations := runtime.limit[operationMutate].fastest[key] + // Seven of the eight fold into the first, so the batch is a batch of one. + if generations == nil || generations[0][0] == 0 || generations[0][sizeClass(8, 0)] != 0 { + t.Fatal("folded mutations counted toward the batch size", generations) + } +} + +func TestALeftOutMutationIsNotPassedByLaterOnesOfItsResource(t *testing.T) { + adapter, runtime, finish := queueBehindBlocker(t, nil) + session := runtime.NewSession() + tickets := []*Ticket{ + submit(t, runtime, foldingPlan("x"), session), + submit(t, runtime, mutationPlan("x"), runtime.NewSession()), + submit(t, runtime, foldingPlan("x"), session), + } + finish() + for _, ticket := range tickets { + wait(t, ticket) + } + want := [][]string{{"blocker"}, {"x"}, {"x"}, {"x"}} + if got := adapter.executed(); !slices.EqualFunc(got, want, slices.Equal) { + t.Fatal("a later mutation passed the one left out", got) + } + if adapter.batch(2)[0].Folds { + t.Fatal("the session's second mutation passed the other session's earlier one") + } +} + func TestCanceledQueuedMutationKeepsSessionOrder(t *testing.T) { adapter := &fakeAdapter{started: make(chan []*execution.Plan, 8)} release := make(chan struct{})