diff --git a/benchmark_runner/redisearch_test.go b/benchmark_runner/redisearch_test.go index c5db359..861dd4b 100644 --- a/benchmark_runner/redisearch_test.go +++ b/benchmark_runner/redisearch_test.go @@ -2,6 +2,7 @@ package benchmark_runner import ( "encoding/json" + "fmt" "os" "os/exec" "strings" @@ -671,3 +672,56 @@ func TestFTSBLatencyCapHighOptIn(t *testing.T) { t.Errorf("Expected q100 >= 1500 ms with --max-latency-seconds=60 (DEBUG SLEEP 2 should record ~2000 ms), got %.2f ms", q100) } } + +// #118: a single failing command in a pipeline window must count as exactly ONE +// error, not `pipeline` errors. Pre-fix, ftsb attributed the pipeline's single +// aggregate error to every command in the window, so one WRONGTYPE reply in a +// window of 10 inflated Errors to 10. +// +// The input is self-contained and deterministic under --workers 1: window 0 +// (rows 0-9) does `SET strkey ...` plus 9 good HSETs (all succeed), and window 1 +// (rows 10-19) has exactly one `HSET strkey ...` (WRONGTYPE, strkey is a string) +// among 9 good HSETs. Correct accounting -> Errors == 1; pre-fix -> Errors == 10. +func TestFTSBPipelineErrorCountedPerCommandNotPerWindow(t *testing.T) { + startRedisContainer(t) + + var b strings.Builder + // Window 0: seed strkey as a STRING, then 9 good hashes -> all succeed. + b.WriteString("WRITE,q,1,SET,strkey,x\n") + for i := 0; i < 9; i++ { + fmt.Fprintf(&b, "WRITE,q,1,HSET,good:%d,f,v\n", i) + } + // Window 1: 5 good, 1 WRONGTYPE (HSET on the string key), 4 good. + for i := 9; i < 14; i++ { + fmt.Fprintf(&b, "WRITE,q,1,HSET,good:%d,f,v\n", i) + } + b.WriteString("WRITE,q,1,HSET,strkey,f,v\n") // WRONGTYPE: strkey is a string + for i := 14; i < 18; i++ { + fmt.Fprintf(&b, "WRITE,q,1,HSET,good:%d,f,v\n", i) + } + + csvPath := "../testdata/pipeline_error_input.csv" + if err := os.WriteFile(csvPath, []byte(b.String()), 0644); err != nil { + t.Fatalf("write input csv: %v", err) + } + t.Cleanup(func() { os.Remove(csvPath) }) + + data := runFTSBReadJSON(t, "../testdata/results.pipeline_error.json", + "--input", csvPath, "--workers", "1", "--pipeline", "10") + + var parsed struct { + Totals struct { + TotalOps int `json:"TotalOps"` + Errors float64 `json:"Errors"` + } `json:"Totals"` + } + if err := json.Unmarshal(data, &parsed); err != nil { + t.Fatalf("parse json: %v", err) + } + if parsed.Totals.TotalOps != 20 { + t.Errorf("TotalOps = %d, want 20", parsed.Totals.TotalOps) + } + if parsed.Totals.Errors != 1 { + t.Errorf("Errors = %v, want 1 (one WRONGTYPE in a pipeline of 10 must not inflate to 10)", parsed.Totals.Errors) + } +} diff --git a/cmd/ftsb_redisearch/cmd_processor.go b/cmd/ftsb_redisearch/cmd_processor.go index d1dfd72..af3e550 100644 --- a/cmd/ftsb_redisearch/cmd_processor.go +++ b/cmd/ftsb_redisearch/cmd_processor.go @@ -378,8 +378,11 @@ func flushPending(p *processor, client radix.Client, pending []pendingCmd) ([]pe hadError := false // Build the action BEFORE timing so the latency window covers only the - // round-trip, not client-side slice bookkeeping. + // round-trip, not client-side slice bookkeeping. For a batch we use + // pipelineErrs (not radix.Pipeline) so we can attribute failures to the + // exact commands that failed instead of blaming the whole window (#118). var action radix.Action + var pe *pipelineErrs if len(pending) == 1 { action = pending[0].action // no need to pipeline a single command } else { @@ -387,16 +390,25 @@ func flushPending(p *processor, client radix.Client, pending []pendingCmd) ([]pe for i := range pending { actions[i] = pending[i].action } - action = radix.Pipeline(actions...) + pe = &pipelineErrs{cmds: actions, errs: make([]error, len(actions))} + action = pe } sendT := time.Now() err := client.Do(action) endT := time.Now() - isTimeout := false if err != nil { hadError = true - isTimeout = logFlushError(pending, err) + logFlushError(pending, err) + } + + // cmdErr returns command i's own error: for a batch that's pe.errs[i] (only + // the commands that actually failed), for a single command it's the Do error. + cmdErr := func(i int) error { + if pe != nil { + return pe.errs[i] + } + return err } // A pipeline is one client round-trip for the whole batch, so attribute the @@ -409,10 +421,15 @@ func flushPending(p *processor, client radix.Client, pending []pendingCmd) ([]pe took := flooredMicros(endT.Sub(sendT)) for i := range pending { pc := &pending[i] + // Each command records its OWN error and timeout status (#118): a single + // bad reply in a pipeline must not mark its siblings as errored. + thisErr := cmdErr(i) + cmdHadError := thisErr != nil + cmdTimeout := cmdHadError && strings.Contains(thisErr.Error(), "i/o timeout") // AddEntry takes (..., rx, tx): received bytes, then sent bytes. Each // command records its OWN counts and labels. rxBytesCount := getRxLen(pc.reply) - stat := benchmark_runner.NewStat().AddEntry([]byte(pc.cmdType), []byte(pc.cmdQueryId), uint64(sendT.Unix()), took, hadError, isTimeout, rxBytesCount, pc.txBytes) + stat := benchmark_runner.NewStat().AddEntry([]byte(pc.cmdType), []byte(pc.cmdQueryId), uint64(sendT.Unix()), took, cmdHadError, cmdTimeout, rxBytesCount, pc.txBytes) p.cmdChan <- *stat } diff --git a/cmd/ftsb_redisearch/pipeline_action.go b/cmd/ftsb_redisearch/pipeline_action.go new file mode 100644 index 0000000..9c81ffe --- /dev/null +++ b/cmd/ftsb_redisearch/pipeline_action.go @@ -0,0 +1,78 @@ +package main + +import ( + "io" + + "github.com/mediocregopher/radix/v3" +) + +// pipelineErrs runs a batch of commands as a single pipeline (one buffered write +// of every command, one batched read of every reply) but, unlike radix.Pipeline, +// it decodes EVERY reply and records each command's individual error instead of +// stopping at the first one. +// +// radix.Pipeline's Run stops decoding at the first failing command and DRAINS +// (discards) the remaining replies, returning a single error for the whole +// batch. ftsb previously attributed that one error to every command in the +// window, so a single WRONGTYPE/OOM reply inflated the error count by up to +// `pipeline` per occurrence (issue #118). Here errs[i] is populated iff command +// i's reply was an error, so accounting is per-command exact. +// +// The write is still a single flush of all commands (via multiMarshal), so this +// keeps the pipelining benefit; the read cost is identical to radix.Pipeline, +// which also decodes/drains all N replies. +type pipelineErrs struct { + cmds []radix.CmdAction + errs []error // len(cmds); errs[i] != nil iff command i failed. +} + +// multiMarshal marshals a batch of CmdActions into one RESP write so the whole +// pipeline is sent in a single flush (mirrors radix.Pipeline's own encode path). +type multiMarshal []radix.CmdAction + +func (m multiMarshal) MarshalRESP(w io.Writer) error { + for _, cmd := range m { + if err := cmd.MarshalRESP(w); err != nil { + return err + } + } + return nil +} + +// Keys returns the union of the batch's keys (radix uses this only for cluster +// routing; ftsb pins a whole batch to one connection so this is informational). +func (p *pipelineErrs) Keys() []string { + var keys []string + for _, cmd := range p.cmds { + keys = append(keys, cmd.Keys()...) + } + return keys +} + +func (p *pipelineErrs) Run(c radix.Conn) error { + // One buffered write of the whole batch. + if err := c.Encode(multiMarshal(p.cmds)); err != nil { + // A write failure breaks the connection, so no reply is coming for any + // command in the batch: every command failed. + for i := range p.cmds { + p.errs[i] = err + } + return err + } + // One batched read: decode every reply so per-command errors are captured. + // A RESP error (e.g. WRONGTYPE) fails only its own command and leaves the + // connection healthy for the rest. A transport error (e.g. i/o timeout) + // breaks the connection, so that command and every command after it fail; + // the ones already decoded succeeded. Either way errs reflects reality + // per-command instead of blaming the whole window. + var first error + for i, cmd := range p.cmds { + if err := c.Decode(cmd); err != nil { + p.errs[i] = err + if first == nil { + first = err + } + } + } + return first +}