qkc/slave 04: implement baseConn and XshardConn compatibility layer

[iteyelmp]

This PR introduces the foundational connection layer for cluster communication.

The main goal is to reproduce the Python cluster connection behavior in Go,
while keeping business logic migration out of scope.

Implemented

baseConn

Introduces the common connection abstraction used by future cluster
communication components.

Implemented:

• frame send/receive lifecycle management
• RPC request/response framework
• RPC ID tracking and validation
• pending RPC management
• connection close propagation
• handler dispatch and error handling
• OpSerializer based payload serialization

XshardConn

Implements slave-to-slave communication on top of baseConn.

Implemented:

• 0-byte metadata transport for slave-slave connections
• PING/PONG identity exchange
• peer identity validation
• shard list validation
• xshard transaction request interfaces
• protocol compatibility stubs for unported business handlers

XshardPool

Adds connection management for xshard peers.

Implemented:

• shard-indexed connection tracking
• inbound connection handling
• duplicate slave detection
• connection cleanup and shutdown handling

[qzhodl]

Please update the PR description to state explicitly whether a full Codex 5.6 review was run against the complete diff before submission, for example: Full Codex 5.6 review completed: yes/no.

My first Codex review of this PR found multiple correctness and lifecycle issues that the existing tests did not expose, including successful acknowledgements for discarded xshard data, unchecked response error codes, stale multi-shard routes, and timeout/late-response teardown. Please run a complete Codex review before submitting code, address or explicitly document its findings, and disclose the review status and any intentional incomplete stubs in the PR description.

[qzhodl]

[P2] On context timeout, the deferred cleanup removes this RPC from pending; a later valid response then reaches the unknown-RPC path in readLoop and closes an otherwise healthy connection. Python retains the map entry until the response or connection close and ignores delivery to a cancelled future. In Go, please retain a bounded cancelled-RPC tombstone until a late response is consumed (or close the connection deliberately on timeout), and add a slow-response regression test.

[qzhodl]

baseConn is explicitly intended to be shared by XshardConn and the future MasterConn, so putting the generic transport/RPC engine under slave gives it the wrong ownership boundary. Please keep wire limited to framing and schemas, move the generic connection layer to a cluster-level package such as qkc/cluster/conn or qkc/cluster/protocol, and keep XshardConn/XshardPool under slave.

[qzhodl]

XshardConn.stateMu shadows the embedded baseConn.stateMu, although the two mutexes protect unrelated state. I do not see a current lock cycle here, but this makes future locking mistakes unnecessarily easy. Please rename this mutex to identityMu and document that it protects remoteID and remoteFullShardIDList.

[syntrust]

rpc_id comes from an atomic counter, but the frame is written later under a different lock (transport.writeMu), so two senders can put N+1 on the wire before N. The peer's monotonic validator rejects it and tears down the connection. Python is immune — no await between self.rpc_id += 1 and write_command on a single-threaded loop. Reachable in normal operation: VerifyAndAddToShards indexes one connection under several shards. Fix: allocate the id inside the write critical section. Suggested test:

func TestSendRPC_ConcurrentSendsPreserveRPCIDOrder(t *testing.T) {
	client, server, cleanup := newTestConnPair(t)
	defer cleanup()
	server.Start()
	client.Start()

	payload, err := serialize.SerializeToBytes(&wire.PingRequest{
		ID: []byte("client"), FullShardIDList: []uint32{0x00010001},
	})
	if err != nil {
		t.Fatal(err)
	}

	const n = 200
	var wg sync.WaitGroup
	start := make(chan struct{})
	for i := 0; i < n; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			<-start
			ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
			defer cancel()
			client.SendRPC(ctx, byte(wire.ClusterOpPing), payload)
		}()
	}
	close(start)
	wg.Wait()

	select {
	case <-server.WaitUntilClosed():
		t.Fatalf("connection torn down: %d concurrent RPCs emitted rpc_ids out of order", n)
	case <-time.After(500 * time.Millisecond):
	}
}

[qzhodl]

OutboundSize is misleading in two ways: after WatchAndIndex, conns contains both outbound and verified inbound connections, and summing the per-shard slices counts route entries rather than unique connections, so one connection serving N shards is counted N times. Please either rename this metric to IndexedRouteCount/RouteCount, or deduplicate connection pointers and expose a true ConnectionCount. If direction-specific counts are required, the pool must retain direction metadata instead of discarding it after indexing. The Targets comment should likewise describe routable shard IDs rather than outbound targets.