From 473dc7e210d398b3baf8c9bf06c88216d2ef304e Mon Sep 17 00:00:00 2001 From: fcostaoliveira Date: Mon, 13 Jul 2026 10:50:15 +0100 Subject: [PATCH 1/3] Fix swapped TX/RX byte counters so TxBytes/RxBytes are labeled correctly AddEntry takes (..., rx, tx), but the single call site passed (txBytesCount, rxBytesCount), so sent bytes were stored as rx and received bytes as tx. Downstream, TxBytes summed Tx() and RxBytes summed Rx(), leaving both metrics mirror-labeled. Because replies are discarded (rcv is a nil interface, getRxLen returns 0), the net effect was that TxBytes always reported 0 and RxBytes reported the sent-byte total. Swap the two arguments at the call site so received bytes land in Rx() and sent bytes in Rx()'s counterpart Tx(), and correct the duplicated struct comment. Tests: - benchmark_runner: AddEntry/NewCmdStat map (rx, tx) into Rx()/Tx() in order. - cmd/ftsb_redisearch: sendFlatCmd records the sent-byte count as Tx() (via a fake radix.Client); verified this test fails on the pre-fix argument order. - getRxLen sizing for string/[]string/nil/other. Closes #111 --- benchmark_runner/stat.go | 4 +- benchmark_runner/stat_test.go | 34 ++++++++++++++ cmd/ftsb_redisearch/cmd_processor.go | 4 +- cmd/ftsb_redisearch/send_stats_test.go | 61 ++++++++++++++++++++++++++ 4 files changed, 100 insertions(+), 3 deletions(-) create mode 100644 benchmark_runner/stat_test.go create mode 100644 cmd/ftsb_redisearch/send_stats_test.go diff --git a/benchmark_runner/stat.go b/benchmark_runner/stat.go index f92154a..794a028 100644 --- a/benchmark_runner/stat.go +++ b/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 { diff --git a/benchmark_runner/stat_test.go b/benchmark_runner/stat_test.go new file mode 100644 index 0000000..1f9c755 --- /dev/null +++ b/benchmark_runner/stat_test.go @@ -0,0 +1,34 @@ +package benchmark_runner + +import "testing" + +// 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()) + } +} diff --git a/cmd/ftsb_redisearch/cmd_processor.go b/cmd/ftsb_redisearch/cmd_processor.go index d02e9aa..42b3625 100644 --- a/cmd/ftsb_redisearch/cmd_processor.go +++ b/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 diff --git a/cmd/ftsb_redisearch/send_stats_test.go b/cmd/ftsb_redisearch/send_stats_test.go new file mode 100644 index 0000000..00ba0fa --- /dev/null +++ b/cmd/ftsb_redisearch/send_stats_test.go @@ -0,0 +1,61 @@ +package main + +import ( + "testing" + "time" + + "github.com/RediSearch/ftsb/benchmark_runner" + radix "github.com/mediocregopher/radix/v3" +) + +// fakeClient is a radix.Client that records nothing and always succeeds, so the +// recording path in sendFlatCmd/sendIfRequired can be exercised without Redis. +type fakeClient struct{ calls int } + +func (f *fakeClient) Do(a radix.Action) error { f.calls++; return nil } +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) { + 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) + } +} From 89ae45308e3810ce7f0888350449354c8ac5ec00 Mon Sep 17 00:00:00 2001 From: fcostaoliveira Date: Mon, 13 Jul 2026 11:01:04 +0100 Subject: [PATCH 2/3] test: cover sendIfRequired error/timeout byte-accounting paths Add fake-radix.Client tests for the error and i/o-timeout branches of sendIfRequired, asserting the sent-byte count is still recorded in Tx() and the error/timeout flags are set. Raises sendIfRequired coverage 58.8% -> 88.2%. --- cmd/ftsb_redisearch/send_stats_test.go | 73 ++++++++++++++++++++++++-- 1 file changed, 69 insertions(+), 4 deletions(-) diff --git a/cmd/ftsb_redisearch/send_stats_test.go b/cmd/ftsb_redisearch/send_stats_test.go index 00ba0fa..113cc78 100644 --- a/cmd/ftsb_redisearch/send_stats_test.go +++ b/cmd/ftsb_redisearch/send_stats_test.go @@ -1,6 +1,7 @@ package main import ( + "errors" "testing" "time" @@ -8,11 +9,15 @@ import ( radix "github.com/mediocregopher/radix/v3" ) -// fakeClient is a radix.Client that records nothing and always succeeds, so the -// recording path in sendFlatCmd/sendIfRequired can be exercised without Redis. -type fakeClient struct{ calls int } +// 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 nil } +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) @@ -59,3 +64,63 @@ func TestGetRxLen(t *testing.T) { 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") + } +} From 36a01f437143ac637ad864dea388b454aa99c147 Mon Sep 17 00:00:00 2001 From: fcostaoliveira Date: Mon, 13 Jul 2026 11:14:01 +0100 Subject: [PATCH 3/3] test: guard user-visible TxBytes/RxBytes mapping + harden byte tests Review-driven hardening (no behavior change): - Add TestGetTotalsMapMapsTxRxCorrectly: guards the GetTotalsMap aggregation labels (txTotalBytes -> "TxBytes", rxTotalBytes -> "RxBytes") -- the JSON the issue #111 symptom appeared in, previously untested. - TestSendFlatCmdRecordsSentBytesAsTx now sets pipeline=1 explicitly (with save/restore) so a stray package global can't make the channel receive block. - TestSendFlatCmdMarksTimeout now also asserts Tx() so it guards byte accounting on the timeout path, not just the timeout flag. - Document that bytelen is an application-payload proxy (excludes RESP framing), accurate for large payloads, approximate for many-tiny-arg commands. --- benchmark_runner/stat_test.go | 33 +++++++++++++++++++++++++- cmd/ftsb_redisearch/cmd_processor.go | 9 +++++-- cmd/ftsb_redisearch/send_stats_test.go | 10 ++++++++ 3 files changed, 49 insertions(+), 3 deletions(-) diff --git a/benchmark_runner/stat_test.go b/benchmark_runner/stat_test.go index 1f9c755..2994a78 100644 --- a/benchmark_runner/stat_test.go +++ b/benchmark_runner/stat_test.go @@ -1,6 +1,10 @@ package benchmark_runner -import "testing" +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 @@ -32,3 +36,30 @@ func TestNewCmdStatMapsRxTxInOrder(t *testing.T) { 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) + } +} diff --git a/cmd/ftsb_redisearch/cmd_processor.go b/cmd/ftsb_redisearch/cmd_processor.go index 42b3625..4336f95 100644 --- a/cmd/ftsb_redisearch/cmd_processor.go +++ b/cmd/ftsb_redisearch/cmd_processor.go @@ -388,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) diff --git a/cmd/ftsb_redisearch/send_stats_test.go b/cmd/ftsb_redisearch/send_stats_test.go index 113cc78..cbd3e09 100644 --- a/cmd/ftsb_redisearch/send_stats_test.go +++ b/cmd/ftsb_redisearch/send_stats_test.go @@ -24,6 +24,13 @@ func (f *fakeClient) Close() error { return nil } // 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 @@ -123,4 +130,7 @@ func TestSendFlatCmdMarksTimeout(t *testing.T) { 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) + } }