nfs/internal/nfs4server/cb.go

// 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
}