Fix swapped TX/RX byte counters (TxBytes/RxBytes mislabeled)

benchmark_runner/stat.go

@@ -22,8 +22,8 @@ type CmdStat struct {
 	latency       uint64 // microseconds latency
 	error         bool
 	timedOut      bool
-	rx            uint64 // bytes received
-	tx            uint64 // bytes received
+	rx            uint64 // bytes received (from Redis replies)
+	tx            uint64 // bytes sent (request/command bytes)
 }
 
 func (c *CmdStat) StartTs() uint64 {

benchmark_runner/stat_test.go

@@ -0,0 +1,65 @@
+package benchmark_runner
+
+import (
+	"testing"
+
+	hdrhistogram "github.com/HdrHistogram/hdrhistogram-go"
+)
+
+// AddEntry's last two params are (rx, tx) in that order. Lock the mapping so a
+// received-byte count lands in Rx() and a sent-byte count lands in Tx(); a
+// regression here silently mislabels the TxBytes/RxBytes throughput metrics.
+func TestAddEntryMapsRxTxInOrder(t *testing.T) {
+	const rxBytes = uint64(7)  // bytes received (reply)
+	const txBytes = uint64(13) // bytes sent (request)
+
+	s := NewStat().AddEntry([]byte("READ"), []byte("q1"), 1000, 42, false, false, rxBytes, txBytes)
+
+	entries := s.CmdStats()
+	if len(entries) != 1 {
+		t.Fatalf("expected 1 entry, got %d", len(entries))
+	}
+	if got := entries[0].Rx(); got != rxBytes {
+		t.Fatalf("Rx() = %d, want %d (received bytes)", got, rxBytes)
+	}
+	if got := entries[0].Tx(); got != txBytes {
+		t.Fatalf("Tx() = %d, want %d (sent bytes)", got, txBytes)
+	}
+}
+
+func TestNewCmdStatMapsRxTxInOrder(t *testing.T) {
+	c := NewCmdStat([]byte("READ"), []byte("q1"), 42, false, false, 7, 13)
+	if c.Rx() != 7 {
+		t.Fatalf("Rx() = %d, want 7", c.Rx())
+	}
+	if c.Tx() != 13 {
+		t.Fatalf("Tx() = %d, want 13", c.Tx())
+	}
+}
+
+// Guards the user-visible mapping in GetTotalsMap: the accumulated sent bytes
+// (txTotalBytes) must surface under "TxBytes" and received under "RxBytes".
+// This is the JSON the #111 symptom appeared in, and the aggregation label is
+// otherwise untested.
+func TestGetTotalsMapMapsTxRxCorrectly(t *testing.T) {
+	h := func() *hdrhistogram.Histogram { return hdrhistogram.New(1, 1_000_000_000, 3) }
+	b := &BenchmarkRunner{
+		totalHistogram:      h(),
+		setupWriteHistogram: h(),
+		writeHistogram:      h(),
+		readHistogram:       h(),
+		readCursorHistogram: h(),
+		updateHistogram:     h(),
+		deleteHistogram:     h(),
+		txTotalBytes:        13, // bytes sent
+		rxTotalBytes:        7,  // bytes received
+	}
+
+	configs := b.GetTotalsMap()
+	if got := configs["TxBytes"]; got != uint64(13) {
+		t.Fatalf("configs[\"TxBytes\"] = %v, want 13 (sent bytes)", got)
+	}
+	if got := configs["RxBytes"]; got != uint64(7) {
+		t.Fatalf("configs[\"RxBytes\"] = %v, want 7 (received bytes)", got)
+	}
+}

cmd/ftsb_redisearch/cmd_processor.go

@@ -302,7 +302,9 @@ func sendIfRequired(p *processor, client radix.Client, cmdType string, cmdQueryI
 			took := uint64(duration.Microseconds())
 			rcv := replies[pos]
 			rxBytesCount += getRxLen(rcv)
-			stat := benchmark_runner.NewStat().AddEntry([]byte(cmdType), []byte(cmdQueryId), uint64(t.Unix()), took, hadError, isTimeout, txBytesCount, rxBytesCount)
+			// AddEntry takes (..., rx, tx): received bytes first, then sent.
+			// txBytesCount is what we send to Redis, rxBytesCount is the reply.
+			stat := benchmark_runner.NewStat().AddEntry([]byte(cmdType), []byte(cmdQueryId), uint64(t.Unix()), took, hadError, isTimeout, rxBytesCount, txBytesCount)
 			p.cmdChan <- *stat
 		}
 		cmds = nil
@@ -386,8 +388,13 @@ func preProcessCmd(row string) (cmdType string, cmdQueryId string, keyPos int, c
 		if initialPos >= 0 {
 			clusterSlot = int(radix.ClusterSlot([]byte(key)))
 		}
-		// Subtract the base64 shrink so byte accounting reflects the decoded
-		// bytes actually sent to Redis, not the larger base64 text in the row.
+		// bytelen approximates the sent (TX) payload: the row minus the leading
+		// cmdType label, minus the base64 shrink (so it reflects decoded bytes,
+		// not the larger base64 text). It is an application-payload proxy, not
+		// exact RESP wire bytes — it still counts CSV separators / the queryId
+		// and pos columns and omits RESP framing (*N\r\n, per-arg $len\r\n). The
+		// error is negligible for large payloads (e.g. vector blobs) but can be
+		// sizable for many-tiny-arg commands.
 		bytelen = uint64(len(row)) - uint64(len(cmdType)) - shrink
 	} else {
 		err = fmt.Errorf("input string does not have the minimum required size of 2: %s", row)

cmd/ftsb_redisearch/send_stats_test.go

@@ -0,0 +1,136 @@
+package main
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/RediSearch/ftsb/benchmark_runner"
+	radix "github.com/mediocregopher/radix/v3"
+)
+
+// fakeClient is a radix.Client whose Do returns a canned error (nil = success),
+// so the recording path in sendFlatCmd/sendIfRequired can be exercised without
+// a real Redis.
+type fakeClient struct {
+	calls int
+	err   error
+}
+
+func (f *fakeClient) Do(a radix.Action) error { f.calls++; return f.err }
+func (f *fakeClient) Close() error            { return nil }
+
+// Regression guard for issue #111: the bytes we SEND to Redis (txBytesCount)
+// must be recorded as Tx(), not Rx(). Before the fix the AddEntry arguments
+// were swapped, so sent bytes were reported under RxBytes and TxBytes was 0.
+func TestSendFlatCmdRecordsSentBytesAsTx(t *testing.T) {
+	// pipeline=1 forces sendIfRequired to flush on the first command. Set it
+	// explicitly (rather than trusting the flag default) so a stray global
+	// left by another test can't make the <-p.cmdChan receive block forever.
+	savedPipeline := pipeline
+	pipeline = 1
+	defer func() { pipeline = savedPipeline }()
+
+	p := &processor{cmdChan: make(chan benchmark_runner.Stat, 1)}
+	const txBytesCount = uint64(4096) // request/sent bytes for this command
+
+	_, _, hadError := sendFlatCmd(
+		p, &fakeClient{}, "WRITE", "w1", "HSET",
+		[]string{"doc:1", "vec", "payload"}, txBytesCount,
+		make([]radix.CmdAction, 0), make([]interface{}, 0), make([]time.Time, 0),
+	)
+	if hadError {
+		t.Fatal("unexpected error from fake client")
+	}
+
+	stat := <-p.cmdChan
+	entries := stat.CmdStats()
+	if len(entries) != 1 {
+		t.Fatalf("expected 1 stat entry, got %d", len(entries))
+	}
+	if got := entries[0].Tx(); got != txBytesCount {
+		t.Fatalf("Tx() = %d, want %d (sent bytes must land in Tx, not Rx)", got, txBytesCount)
+	}
+	// Replies are discarded (rcv is a nil interface), so received bytes are 0.
+	if got := entries[0].Rx(); got != 0 {
+		t.Fatalf("Rx() = %d, want 0 (replies are not captured)", got)
+	}
+}
+
+func TestGetRxLen(t *testing.T) {
+	if got := getRxLen("abc"); got != 3 {
+		t.Fatalf("getRxLen(string) = %d, want 3", got)
+	}
+	if got := getRxLen([]string{"ab", "cde"}); got != 5 {
+		t.Fatalf("getRxLen([]string) = %d, want 5", got)
+	}
+	if got := getRxLen(nil); got != 0 {
+		t.Fatalf("getRxLen(nil) = %d, want 0", got)
+	}
+	if got := getRxLen(42); got != 0 {
+		t.Fatalf("getRxLen(int) = %d, want 0", got)
+	}
+}
+
+// On a command error (continue-on-error), a stat must still be recorded, marked
+// as an error, and carry the correct sent-byte count in Tx().
+func TestSendFlatCmdRecordsErrorStatWithCorrectTx(t *testing.T) {
+	savedPipeline, savedContinue := pipeline, continueOnErr
+	pipeline, continueOnErr = 1, true
+	defer func() { pipeline, continueOnErr = savedPipeline, savedContinue }()
+
+	p := &processor{cmdChan: make(chan benchmark_runner.Stat, 1)}
+	const txBytesCount = uint64(128)
+
+	_, _, hadError := sendFlatCmd(
+		p, &fakeClient{err: errors.New("connection refused")}, "WRITE", "w1", "HSET",
+		[]string{"doc:1"}, txBytesCount,
+		make([]radix.CmdAction, 0), make([]interface{}, 0), make([]time.Time, 0),
+	)
+	if !hadError {
+		t.Fatal("expected hadError=true when client.Do returns an error")
+	}
+
+	stat := <-p.cmdChan
+	entries := stat.CmdStats()
+	if len(entries) != 1 {
+		t.Fatalf("expected 1 stat entry, got %d", len(entries))
+	}
+	if !entries[0].Error() {
+		t.Fatal("entry should be marked as an error")
+	}
+	if got := entries[0].Tx(); got != txBytesCount {
+		t.Fatalf("Tx() = %d, want %d (sent bytes recorded even on error)", got, txBytesCount)
+	}
+}
+
+// An i/o timeout error must set the timedOut flag on the recorded stat.
+func TestSendFlatCmdMarksTimeout(t *testing.T) {
+	savedPipeline, savedContinue := pipeline, continueOnErr
+	pipeline, continueOnErr = 1, true
+	defer func() { pipeline, continueOnErr = savedPipeline, savedContinue }()
+
+	p := &processor{cmdChan: make(chan benchmark_runner.Stat, 1)}
+	_, _, hadError := sendFlatCmd(
+		p, &fakeClient{err: errors.New("dial tcp 127.0.0.1:6379: i/o timeout")}, "READ", "r1", "FT.SEARCH",
+		[]string{"idx"}, 64,
+		make([]radix.CmdAction, 0), make([]interface{}, 0), make([]time.Time, 0),
+	)
+	if !hadError {
+		t.Fatal("expected hadError=true on timeout")
+	}
+	stat := <-p.cmdChan
+	entries := stat.CmdStats()
+	if len(entries) != 1 {
+		t.Fatalf("expected 1 stat entry, got %d", len(entries))
+	}
+	if !entries[0].TimedOut() {
+		t.Fatal("entry should be marked as timed out for an i/o timeout error")
+	}
+	if !entries[0].Error() {
+		t.Fatal("a timeout is also an error")
+	}
+	if got := entries[0].Tx(); got != 64 {
+		t.Fatalf("Tx() = %d, want 64 (sent bytes recorded even on timeout)", got)
+	}
+}