feat: full NFSv4.2 server and client in pure Go

Assisted-by: GLM 5.3 Flash

internal/nfs4server/cb.go

@@ -0,0 +1,203 @@
+// Copyright (c) 2026 Petr Balvín <opensource@petrbalvin.org> (https://petrbalvin.org)
+// SPDX-License-Identifier: MIT
+
+package nfs4server
+
+import (
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"sourcedock.dev/petrbalvin/nfs/internal/nfs4"
+	"sourcedock.dev/petrbalvin/nfs/internal/rpc"
+)
+
+// cbTimeout bounds one CB call: the client that does not answer in time
+// loses the connection's callback service.
+const cbTimeout = 5 * time.Second
+
+// cbQueue bounds the callback work one connection may hold in flight.
+const cbQueue = 32
+
+// A cbResult is the outcome of one delivered CB_COMPOUND.
+type cbResult struct {
+	res    nfs4.CompoundRes
+	bodies [][]byte
+	err    error
+}
+
+// A cbWork is one queued CB_COMPOUND: the operations, the channel the
+// synchronous caller waits on, and the action to run once delivery
+// ends, whatever its outcome. A nil result channel makes the work fire
+// and forget.
+type cbWork struct {
+	tag      string
+	clientid uint64
+	ops      [][]byte
+	result   chan cbResult
+	done     func(cbResult)
+}
+
+// A connCB is the per connection callback machinery: one worker
+// goroutine delivers every CB_COMPOUND of the connection in order, the
+// write side is shared with the request replies, and the pending map
+// routes CB replies back to the worker. The worker exists so the
+// dispatch path never blocks on a reply only the connection's own read
+// loop can route: a CB issued while serving a request of the same
+// connection would otherwise wait on itself.
+type connCB struct {
+	conn      net.Conn
+	writeMu   *sync.Mutex
+	cbProgram uint32
+
+	mu      sync.Mutex
+	nextXID uint32
+	pending map[uint32]chan []byte
+
+	work chan cbWork
+}
+
+func newConnCB(conn net.Conn, writeMu *sync.Mutex) *connCB {
+	c := &connCB{
+		conn:    conn,
+		writeMu: writeMu,
+		pending: make(map[uint32]chan []byte),
+		work:    make(chan cbWork, cbQueue),
+	}
+	go c.worker()
+	return c
+}
+
+// worker delivers the queued callbacks one at a time, which keeps the
+// back channel slot discipline of one call in flight per slot.
+func (c *connCB) worker() {
+	for w := range c.work {
+		r := c.deliver(w.tag, w.clientid, w.ops)
+		if w.done != nil {
+			w.done(r)
+		}
+		if w.result != nil {
+			w.result <- r
+		}
+	}
+}
+
+// deliver sends one CB_COMPOUND over the connection and returns the
+// decoded result. It runs on the worker goroutine only.
+func (c *connCB) deliver(tag string, clientid uint64, ops [][]byte) cbResult {
+	c.mu.Lock()
+	xid := c.nextXID
+	c.nextXID++
+	ch := make(chan []byte, 1)
+	c.pending[xid] = ch
+	c.mu.Unlock()
+
+	header := rpc.Call{XID: xid, Program: c.program(), Version: nfs4.Version,
+		Procedure: nfs4.ProcCompound, Cred: rpc.AuthNull}
+	req, err := rpc.AppendCall(nil, header)
+	if err != nil {
+		c.dropPending(xid)
+		return cbResult{err: err}
+	}
+	args := nfs4.AppendCBCompoundArgs(nil, tag, nfs4.MinorVersion, 0, ops)
+	if err := c.write(append(req, args...)); err != nil {
+		c.dropPending(xid)
+		return cbResult{err: err}
+	}
+
+	var rec []byte
+	select {
+	case rec = <-ch:
+	case <-time.After(cbTimeout):
+		c.dropPending(xid)
+		return cbResult{err: fmt.Errorf("cb: no reply in %s", cbTimeout)}
+	}
+
+	reply, err := rpc.DecodeReply(rec)
+	if err != nil {
+		return cbResult{err: err}
+	}
+	if reply.Rejected || reply.Status != rpc.AcceptSuccess {
+		return cbResult{err: fmt.Errorf("cb: rpc status %d, rejected %v", reply.Status, reply.Rejected)}
+	}
+	r, bodies, err := nfs4.DecodeCompoundResBodiesCB(reply.Body)
+	return cbResult{res: r, bodies: bodies, err: err}
+}
+
+// tryQueue enqueues one work item, or fails when the queue is full.
+func (c *connCB) tryQueue(w cbWork) error {
+	select {
+	case c.work <- w:
+		return nil
+	default:
+		return fmt.Errorf("cb: the callback queue is full")
+	}
+}
+
+// post queues one fire and forget CB_COMPOUND with an action to run
+// once delivery ends.
+func (c *connCB) post(tag string, clientid uint64, ops [][]byte, done func(cbResult)) error {
+	return c.tryQueue(cbWork{tag: tag, clientid: clientid, ops: ops, done: done})
+}
+
+// setProgram binds the callback program negotiated by CREATE_SESSION.
+func (c *connCB) setProgram(program uint32) {
+	c.mu.Lock()
+	c.cbProgram = program
+	c.mu.Unlock()
+}
+
+// program reports the bound callback program.
+func (c *connCB) program() uint32 {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.cbProgram
+}
+
+// route delivers a record the demultiplexer identified as a reply to one
+// of this connection's CB calls.
+func (c *connCB) route(rec []byte) bool {
+	xid, _, err := rpc.PeekHeader(rec)
+	if err != nil {
+		return false
+	}
+	c.mu.Lock()
+	ch, ok := c.pending[xid]
+	if ok {
+		delete(c.pending, xid)
+	}
+	c.mu.Unlock()
+	if !ok {
+		return false
+	}
+	ch <- rec
+	return true
+}
+
+func (c *connCB) dropPending(xid uint32) {
+	c.mu.Lock()
+	delete(c.pending, xid)
+	c.mu.Unlock()
+}
+
+// wirePool recycles the record marking buffers of outgoing replies. The
+// buffer lives only for the write: the connection copies it to the
+// kernel, nothing retains a reference, and the cap keeps one oversized
+// reply from pinning its memory forever.
+var wirePool = sync.Pool{New: func() any { b := make([]byte, 0, 16<<10); return b }}
+
+// maxWireBuf bounds the buffers the pool holds on to.
+const maxWireBuf = maxRecord + 16
+
+func (c *connCB) write(data []byte) error {
+	c.writeMu.Lock()
+	defer c.writeMu.Unlock()
+	buf := wirePool.Get().([]byte)[:0]
+	buf = rpc.AppendRecord(buf, data)
+	_, err := c.conn.Write(buf)
+	if cap(buf) <= maxWireBuf {
+		wirePool.Put(buf)
+	}
+	return err
+}