TimeoutDelay returns a desynchronized connection to the pool: later commands silently receive the previous command's response

Summary

When BasePolicy.TimeoutDelay > 0 and a single-key command times out before the 8-byte proto header arrives, singleCommand.salvageConn discards nothing and returns the connection to the pool with the entire unread response still in the socket. Every subsequent command that picks up that connection reads the previous command's response and returns it to the caller as its own record — no error, no timeout, just the wrong data, and the connection stays shifted by one response.

A read timeout before the header arrives is the ordinary case, not an exotic one: it is what a single lost response packet looks like when SocketTimeout is shorter than the TCP retransmission timeout.

With TimeoutDelay = 0 the same scenario is handled correctly — the connection is closed.

Environment

• github.com/aerospike/aerospike-client-go/v8 v8.8.0
• Go 1.26.0, linux/amd64
• Aerospike Community Edition 6.1.0.3 (the bug is entirely client-side)

salvageConn at the current tip of the v8 branch (d68cca2e, 2026-08-27) is byte-identical to v8.8.0, so this is not something already fixed since the release.

Reproduction

Self-contained, no fault injection inside the client: a TCP proxy sits between the client and the server and delays the server→client direction by 60 ms, while SocketTimeout is 20 ms. main.go is attached below.

docker run -d --name as-repro -p 13000:3000 \
  -e NAMESPACE=test -e MEM_GB=1 aerospike/aerospike-server:6.1.0.3
go mod init repro && go get github.com/aerospike/aerospike-client-go/v8@v8.8.0
go run . 127.0.0.1:13000

The program writes key0=RECORD-0 and key1=RECORD-1, lets one Get(key0) die on its socket deadline, waits for the salvage goroutine to finish, and then performs six ordinary reads alternating between the two keys.

Output:

TimeoutDelay=0 (salvage disabled)
  connections closed by the client: 1, connections-recovered: 0
  PASS: 6/6 subsequent reads returned the right record

TimeoutDelay=300ms (salvage enabled)
  read 0: asked for RECORD-1, server returned "RECORD-0"
  read 1: asked for RECORD-0, server returned "RECORD-1"
  read 2: asked for RECORD-1, server returned "RECORD-0"
  read 3: asked for RECORD-0, server returned "RECORD-1"
  read 4: asked for RECORD-1, server returned "RECORD-0"
  read 5: asked for RECORD-0, server returned "RECORD-1"
  connections closed by the client: 0, connections-recovered: 1
  FAIL: 6 of 6 subsequent reads returned the wrong record

connections-recovered reports 1 for a connection that is in fact corrupted. That counter is incremented in exactly one place in the whole client — the last line of singleCommand.salvageConn — so it also serves as proof that this code path, and not something else, is what ran.

All six reads fail rather than just the one whose key hashes to the same sub-heap, because salvageConn returns the connection through node.PutConnection(conn), which always uses hint 0, and connectionHeap.Poll falls back to a round-robin across every sub-heap.

SendKey = true does not help — the response is parsed as a valid record regardless of which key was requested.

Root cause

1. discardedCount is wrong exactly when the connection needs draining most

single_command.go:85:

discardedCount := int(cmd.receiveSize - conn.totalReceived)

for discardedCount > 0 {
    ...
}

conn.refresh()
node.PutConnection(conn)

For a singleCommand, cmd.receiveSize is assigned in exactly one place, record_parser.go:68, i.e. only after the 8-byte proto header has been read and parsed. If the deadline expires before the header arrives, the field still holds whatever it held before:

• On the first attempt of a fresh command it is the zero value, so discardedCount == 0, the loop never executes, and the connection goes back to the pool untouched, with the full response pending in the socket. This is the case the reproduction exercises.
• cmd.receiveSize is never reset between retry iterations of the same *baseCommand (there is no assignment of 0 to it anywhere in the package, unlike conn.totalReceived, which refresh() clears on every checkout from the pool). So with MaxRetries > 0 it can still hold the length parsed during an earlier iteration that ran on a different connection, and discardedCount becomes a non-zero number unrelated to the socket actually being drained — draining too few bytes, or reaching past the end of the response.

conn.refresh() then resets totalReceived to 0, erasing the only remaining trace that the connection is out of sync.

2. The salvage deadline is never applied to the socket

salvageConn assigns conn.deadline = time.Now().Add(timeoutDelay), but then reads through bufio.NewReader(conn.conn) — the raw net.Conn, bypassing Connection.Read/Write, which are the only callers of updateDeadline(), itself the only caller of conn.SetDeadline(...) (connection.go:231, :261, :339). Note also that only Connection.socketDeadline ever reaches the syscall; Connection.deadline is merely an upper bound consulted by updateDeadline().

Nothing on the failure path resets the socket deadline — not Connection.Read, not errToAerospikeErr, not refresh() — so the socket still carries the already-expired deadline of the command that just failed, and the first reader.Discard returns os.ErrDeadlineExceeded immediately with discarded == 0. The code then takes the discarded < discardedCount branch and closes the connection.

In practice this means the salvage path can never succeed in the one case where discardedCount is correct — when the header was received. In that scenario the measured connections-recovered stays at 0, i.e. TimeoutDelay buys nothing and only costs a goroutine.

Taken together with (1): when the mechanism can act it corrupts the connection, and when it would be correct it cannot act.

3. Data race on the command object

The salvage goroutine is started while the retry loop keeps using the same *baseCommand (command.go:3948-3950):

if cmd.conn.salvageConnection && policy.TimeoutDelay > 0 {
    go ifc.salvageConn(policy.TimeoutDelay, cmd.conn, cmd.node)
    continue
}

Reproduced with -race (same program, MaxRetries = 3):

WARNING: DATA RACE
Write at 0x00c000456038 by main goroutine:
  aerospike-client-go/v8.(*baseCommand).executeAt()
      command.go:3784 +0x532        // cmd.node, err = ifc.getNode(ifc)
  ...
Previous read at 0x00c000456038 by goroutine 55:
  aerospike-client-go/v8.(*singleCommand).salvageConn()
      single_command.go:104 +0x437  // applyConnectionRecoveredMetrics(cmd.node)
  aerospike-client-go/v8.(*baseCommand).executeAt.gowrap2()
      command.go:3950 +0x69
Found 1 data race(s)

Worth noting that salvageConn already receives node as a parameter and ignores it in favour of cmd.node.

cmd.conn is shared the same way and does not even need a retry to race: command.go:3915 assigns cmd.conn = nil in the same iteration, immediately after the go statement on :3908, while the goroutine reads cmd.conn at single_command.go:78 and writes it at :93.

cmd.receiveSize is shared too — read by the goroutine at single_command.go:85, reassigned by the next attempt at record_parser.go:68 — which is the concurrent version of the staleness described in (1).

Comparison with the Java client

The Java implementation, ConnectionRecover, handles the case that is missing here: when the timeout occurred before the header arrived (offset == 0), it reads the header itself and derives the remaining length from it via parseProto. It bounds the work with a budget rather than a single deadline — the socket timeout is set to 1 ms and each SocketTimeoutException compares elapsed time against timeoutDelay, calling abort() once the budget is exhausted — and any Throwable also leads to abort(). It is also explicit about what it will not do: recovering compressed multi-record responses throws rather than guessing.

Impact

Silent wrong data returned to the application, on a released version, from a policy field whose documentation describes it as a background recovery optimization. There is no error to catch and no metric that flags it — connections-recovered increments as if the connection were healthy. Once a connection is shifted it stays shifted until IdleTimeout or a network error retires it, so a single timeout can poison one pooled connection for minutes.

Suggested fix

In singleCommand.salvageConn:

1. Set a real socket deadline bounded by timeoutDelay (go through Connection.Read, or call conn.conn.SetDeadline explicitly) before reading anything.
2. If conn.totalReceived < 8, read the remainder of the proto header first and take the remaining length from it as int(sz & 0xFFFFFFFFFFFF), measured from the end of the 8-byte header, instead of trusting cmd.receiveSize. Note that emptySocket is not the right template here: it subtracts headerLength because it runs at a point where all 30 bytes of _MSG_TOTAL_HEADER_SIZE have already been consumed.
3. Refuse to drain compressed responses. For msgType == 4 the bytes left in the socket are governed by the compressed size, whereas cmd.receiveSize holds the uncompressed length, and draining through the raw net.Conn bypasses the inflater and its limitReader entirely. Closing the connection is the only safe action, which is the position the Java client takes for the multi-record case.
4. Close the connection on any error, short read, or deadline expiry, and do not increment connections-recovered in that case.
5. Do not read or write cmd fields from the salvage goroutine — pass receiveSize, totalReceived and the node in as arguments.

I can prepare a PR along these lines against the v8 branch, with regression tests for both scenarios (timeout before the header, timeout after the header), if that is welcome.

Related, lower priority

TimeoutDelay is effectively a single-command feature today: baseMultiCommand.salvageConn (multi_command.go:520) just closes the connection, and both scanPartitionCommand.salvageConn (scan_partition_command.go:94) and scanPartitionObjectsCommand.salvageConn (scan_partition_objects_command.go:98) carry a // TODO: implement scan and query commands later. That is safe behaviour, but it may be worth saying so in the policy documentation.

Separately, Node.getConnectionWithHint (node.go:562) accepts a timeoutDelay parameter and never uses it.

// Minimal reproduction for a data-corruption bug in BasePolicy.TimeoutDelay
// (singleCommand.salvageConn) in aerospike-client-go/v8.
//
// It puts a TCP proxy between the client and the server. The proxy delays the
// server->client direction so that a read command hits its SocketTimeout before
// the 8-byte proto header arrives. That is what an ordinary lost response packet
// looks like to the client.
//
// With TimeoutDelay > 0 the client hands the connection to salvageConn, which
// computes discardedCount = cmd.receiveSize - conn.totalReceived. Both are 0,
// because cmd.receiveSize is only assigned after the header has been parsed, so
// nothing is discarded and the connection is returned to the pool with the whole
// unread response still sitting in the socket. Every later command on that
// connection reads the previous command's response and returns it as its own,
// with no error.
//
// Run:
//
//	docker run -d --name as-repro -p 13000:3000 \
//	  -e NAMESPACE=test -e MEM_GB=1 aerospike/aerospike-server:6.1.0.3
//	go mod init repro && go get github.com/aerospike/aerospike-client-go/v8@v8.8.0
//	go run . 127.0.0.1:13000
package main

import (
	"fmt"
	"net"
	"os"
	"strconv"
	"sync"
	"sync/atomic"
	"time"

	as "github.com/aerospike/aerospike-client-go/v8"
)

var namespace = "test" // override with AS_NAMESPACE

const (
	setName = "salvage"
	binName = "v"

	socketTimeout = 20 * time.Millisecond
	totalTimeout  = 150 * time.Millisecond
	timeoutDelay  = 300 * time.Millisecond
	responseDelay = 60 * time.Millisecond // > socketTimeout, so the read times out
)

// proxy forwards TCP and can delay the server->client direction on demand.
type proxy struct {
	ln      net.Listener
	target  string
	delayNs atomic.Int64
	closed  atomic.Int64 // connections torn down, i.e. the client closed its side
}

func newProxy(target string) (*proxy, error) {
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		return nil, err
	}
	p := &proxy{ln: ln, target: target}
	go func() {
		for {
			c, err := ln.Accept()
			if err != nil {
				return
			}
			go p.handle(c)
		}
	}()
	return p, nil
}

func (p *proxy) handle(client net.Conn) {
	server, err := net.Dial("tcp", p.target)
	if err != nil {
		client.Close()
		return
	}
	var once sync.Once
	shut := func() { once.Do(func() { client.Close(); server.Close() }) }

	go func() { // client -> server, never delayed
		buf := make([]byte, 64*1024)
		for {
			n, rerr := client.Read(buf)
			if n > 0 {
				if _, werr := server.Write(buf[:n]); werr != nil {
					break
				}
			}
			if rerr != nil {
				p.closed.Add(1)
				break
			}
		}
		shut()
	}()

	buf := make([]byte, 64*1024) // server -> client, delayed on demand
	for {
		n, rerr := server.Read(buf)
		if n > 0 {
			if d := time.Duration(p.delayNs.Load()); d > 0 {
				time.Sleep(d)
			}
			if _, werr := client.Write(buf[:n]); werr != nil {
				break
			}
		}
		if rerr != nil {
			break
		}
	}
	shut()
}

func hostPort(addr string) (string, int) {
	h, ps, _ := net.SplitHostPort(addr)
	p, _ := strconv.Atoi(ps)
	return h, p
}

func run(delay time.Duration) (failures int, err error) {
	target := "127.0.0.1:13000"
	if len(os.Args) > 1 {
		target = os.Args[1]
	}
	p, err := newProxy(target)
	if err != nil {
		return 0, err
	}
	host, port := hostPort(p.ln.Addr().String())

	cp := as.NewClientPolicy()
	cp.Timeout = 5 * time.Second
	cp.ConnectionQueueSize = 4
	cp.MinConnectionsPerNode = 2
	cp.LimitConnectionsToQueueSize = true
	client, aerr := as.NewClientWithPolicyAndHost(cp, as.NewHost(host, port))
	if aerr != nil {
		return 0, aerr
	}
	defer client.Close()
	client.EnableMetrics(as.DefaultMetricsPolicy())

	relaxed := as.NewWritePolicy(0, 0)
	relaxed.SocketTimeout = 3 * time.Second
	relaxed.TotalTimeout = 3 * time.Second
	for i := 0; i < 2; i++ {
		key, _ := as.NewKey(namespace, setName, fmt.Sprintf("key%d", i))
		if aerr := client.PutBins(relaxed, key, as.NewBin(binName, fmt.Sprintf("RECORD-%d", i))); aerr != nil {
			return 0, aerr
		}
	}

	read := as.NewPolicy()
	read.MaxRetries = 0
	read.SocketTimeout = 3 * time.Second
	read.TotalTimeout = 3 * time.Second
	key0, _ := as.NewKey(namespace, setName, "key0")
	key1, _ := as.NewKey(namespace, setName, "key1")
	if _, aerr := client.Get(read, key0); aerr != nil { // warm the pool
		return 0, aerr
	}
	time.Sleep(300 * time.Millisecond)
	closedBefore := p.closed.Load()

	// The command that dies on its socket deadline before the header arrives.
	tight := as.NewPolicy()
	tight.MaxRetries = 0
	tight.SocketTimeout = socketTimeout
	tight.TotalTimeout = totalTimeout
	tight.TimeoutDelay = delay
	p.delayNs.Store(int64(responseDelay))
	_, timeoutErr := client.Get(tight, key0)
	p.delayNs.Store(0)
	if timeoutErr == nil {
		return 0, fmt.Errorf("expected a timeout, got none: raise responseDelay")
	}

	time.Sleep(600 * time.Millisecond) // let the salvage goroutine finish

	// Every later read must return its own record.
	for i := 0; i < 6; i++ {
		key, want := key1, "RECORD-1"
		if i%2 == 1 {
			key, want = key0, "RECORD-0"
		}
		rec, rerr := client.Get(read, key)
		if rerr != nil {
			failures++
			fmt.Printf("  read %d: error: %v\n", i, rerr)
			continue
		}
		got, _ := rec.Bins[binName].(string)
		if got != want {
			failures++
			fmt.Printf("  read %d: asked for %s, server returned %q\n", i, want, got)
		}
	}

	recovered := 0.0
	if stats, serr := client.Stats(); serr == nil {
		if agg, ok := stats["cluster-aggregated-stats"].(map[string]any); ok {
			recovered, _ = agg["connections-recovered"].(float64)
		}
	}
	fmt.Printf("  connections closed by the client: %d, connections-recovered: %.0f\n",
		p.closed.Load()-closedBefore, recovered)
	return failures, nil
}

func main() {
	if ns := os.Getenv("AS_NAMESPACE"); ns != "" {
		namespace = ns
	}
	for _, c := range []struct {
		name  string
		delay time.Duration
	}{
		{"TimeoutDelay=0 (salvage disabled)", 0},
		{"TimeoutDelay=300ms (salvage enabled)", timeoutDelay},
	} {
		fmt.Printf("%s\n", c.name)
		failures, err := run(c.delay)
		if err != nil {
			fmt.Printf("  setup error: %v\n\n", err)
			continue
		}
		if failures == 0 {
			fmt.Printf("  PASS: 6/6 subsequent reads returned the right record\n\n")
		} else {
			fmt.Printf("  FAIL: %d of 6 subsequent reads returned the wrong record\n\n", failures)
		}
	}
}