Count pipeline errors per-command, not per-window

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)
+	}
+}

cmd/ftsb_redisearch/cmd_processor.go

@@ -378,25 +378,37 @@ 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 {
 		actions := make([]radix.CmdAction, len(pending))
 		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
 	}
 

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
+}