// Copyright (c) 2026 Petr Balvín <opensource@petrbalvin.org> (https://petrbalvin.org)
// SPDX-License-Identifier: MIT

// Package nfs4server turns reassembled ONC RPC records into NFSv4.2
// operations against a virtual filesystem: the stateless operations, the
// sessions with their slot table, and the open, lock and delegation state.
package nfs4server

import (
	crand "crypto/rand"
	"crypto/tls"
	"encoding/binary"
	"errors"
	"io/fs"
	"net"
	"strconv"
	"sync"
	"time"

	"sourcedock.dev/petrbalvin/nfs/internal/nfs4"
	"sourcedock.dev/petrbalvin/nfs/internal/nfsfs"
	"sourcedock.dev/petrbalvin/nfs/internal/rpc"
	"sourcedock.dev/petrbalvin/nfs/internal/xdr"
)

// maxRecord bounds one ONC RPC record. A call larger than this is refused
// before its bytes are buffered.
const maxRecord = 4 << 20

// supportedAccess is the ACCESS mask this server answers for.
const supportedAccess = uint32(nfsfs.AccessRead | nfsfs.AccessLookup |
	nfsfs.AccessModify | nfsfs.AccessExtend | nfsfs.AccessDelete | nfsfs.AccessExec)

// The anonymous identity root squash maps a root credential onto: the
// nobody user and group of the classic NFS exports.
const (
	nobodyUID = 65534
	nobodyGID = 65534
)

// A Handler serves one connection at a time from the FS it is given. It is
// the Handle hook of the server package.
type Handler struct {
	FS nfsfs.FS

	// LeasePeriod is how long the server keeps state for a client that
	// stops renewing. Zero means the default of 90 seconds.
	LeasePeriod time.Duration

	// StateDir is the directory the client state persists into: handles
	// and opens survive a restart when it is set.
	StateDir string

	// GracePeriod is the reclaim window after the server start. Zero means
	// the default of 90 seconds.
	GracePeriod time.Duration

	// ServerKey is the long term Kerberos key of the nfs service
	// principal; set together with ServiceName it enables the RPCSEC_GSS
	// procedures of RFC 2203.
	ServerKey   []byte
	ServiceName string

	// TLSConfig, when set, lets the client upgrade the connection to
	// TLS through the AUTH_TLS probe of RFC 9289.
	TLSConfig *tls.Config

	// LogOps answers one log line per operation on stderr: the operation,
	// the status it returned and the time it took. Off by default.
	LogOps bool

	// RootSquash maps the root identity of a client onto nobody: a
	// credential that claims uid 0 acts as uid 65534 with group 65534,
	// so the permission bits of nobody decide and the objects root
	// creates carry nobody. Off by default, which keeps the trust AUTH_SYS
	// hands to the claim.
	RootSquash bool

	// DeviceAddr is the universal address the pNFS data server, which is
	// the metadata server itself, answers on. Empty means the local
	// address of the request connection, with loopback as the last
	// resort.
	DeviceAddr string

	mu          sync.Mutex
	verifier    sync.Once // the boot verifier is generated exactly once
	writeVer    [8]byte
	store       *sessionStore
	states      *stateStore
	lockStore   *lockStore
	delegStore  *delegStore
	layoutStore *layoutServer
	dirDelegSt  *dirDelegStore
	gssSt       *gssStore
	refSt       *referralStore
	nattrSt     *nattrStore
	grace       *grace
	probes      int

	// excl guards and holds the create verifiers of exclusive opens.
	exclMu sync.Mutex
	excl   map[string][8]byte
}

// graced returns the grace window, made once per handler. The verifier
// generation happens outside the handler mutex, because writeVerifier
// takes the same lock.
func (h *Handler) graced() *grace {
	h.mu.Lock()
	g := h.grace
	h.mu.Unlock()
	if g != nil {
		return g
	}
	g = newGrace(h.gracePeriod())
	h.mu.Lock()
	if h.grace == nil {
		h.grace = g
	}
	g = h.grace
	h.mu.Unlock()
	return g
}

// gracePeriod resolves the configured grace period: an explicitly set
// value wins, zero means the default of 90 seconds.
func (h *Handler) gracePeriod() time.Duration {
	if h.GracePeriod != 0 {
		return h.GracePeriod
	}
	return 90 * time.Second
}

// delegs returns the delegation store, made once per handler.
func (h *Handler) delegs() *delegStore {
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.delegStore == nil {
		h.delegStore = newDelegStore()
	}
	return h.delegStore
}

// leasePeriod resolves the configured lease period.
func (h *Handler) leasePeriod() time.Duration {
	if h.LeasePeriod > 0 {
		return h.LeasePeriod
	}
	return 90 * time.Second
}

// locks returns the byte range lock store, made once per handler.
func (h *Handler) locks() *lockStore {
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.lockStore == nil {
		h.lockStore = newLockStore()
	}
	return h.lockStore
}

// openStates returns the OPEN state store, made once per handler.
func (h *Handler) openStates() *stateStore {
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.states == nil {
		h.states = newStateStore(h.StateDir)
	}
	return h.states
}

// layouts returns the pNFS layout store, made once per handler.
func (h *Handler) layouts() *layoutServer {
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.layoutStore == nil {
		h.layoutStore = newLayoutServer()
	}
	return h.layoutStore
}

// sessions returns the session store, made once per handler with a
// random server prefix for its session ids.
func (h *Handler) sessions() *sessionStore {
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.store == nil {
		var prefix [4]byte
		if _, err := crand.Read(prefix[:]); err != nil {
			panic("nfs4server: the random source failed: " + err.Error())
		}
		h.store = newSessionStore(prefix)
	}
	return h.store
}

// writeVerifier returns the server boot verifier of RFC 8881 section
// 8.10: a value that changes when the server restarts, so a client can
// tell that its write replays are meaningless. It is made once, from the
// system's random source.
func (h *Handler) writeVerifier() [8]byte {
	h.verifier.Do(func() {
		if _, err := crand.Read(h.writeVer[:]); err != nil {
			panic("nfs4server: the random source failed: " + err.Error())
		}
	})
	return h.writeVer
}

// writer returns the mutating half of the filesystem, or nil when the
// backend serves reads only.
func (h *Handler) writer() nfsfs.Writer {
	w, _ := h.FS.(nfsfs.Writer)
	return w
}

// a cred holds the identity the client asserted, or the identity of nobody
// when the credential flavour carries no usable claim.
type cred struct {
	uid, gid uint32
	groups   []uint32
}

// the fh register of one COMPOUND: the current and the saved handle,
// the client whose session drives the compound, and the component the
// last LOOKUP resolved.
type fhreg struct {
	cur, saved nfsfs.Handle
	haveCur    bool
	clientID   uint64
	lastLookup string
}

// HandleConn serves ONC RPC calls until the connection closes. Errors on
// the wire end the session; protocol errors are answered and it stays.
// An AUTH_TLS NULL probe upgrades the connection to TLS per RFC 9289
// when the handler carries a TLSConfig.
func (h *Handler) HandleConn(conn net.Conn) {
	defer conn.Close()
	tlsActive := false
	writeMu := &sync.Mutex{}
	ctx := newConnCB(conn, writeMu)
	for {
		rec, err := rpc.ReadRecord(conn, maxRecord)
		if err != nil {
			return
		}
		_, mtype, err := rpc.PeekHeader(rec)
		if err != nil {
			return
		}
		if mtype == rpc.MsgReply {
			// A reply to a CB call this connection issued; the CB caller
			// waits on its xid.
			if !ctx.route(rec) {
				return
			}
			continue
		}
		call, args, err := rpc.DecodeCall(rec)
		if err != nil {
			return
		}
		// The RPC-with-TLS probe: a NULL procedure under AUTH_TLS. The
		// answer carries the STARTTLS token and the connection upgrades
		// to TLS before any further record is read.
		if call.Cred.Flavor == rpc.FlavorTLS {
			if call.Procedure != nfs4.ProcNull || tlsActive || h.TLSConfig == nil {
				reply := rpc.AppendRejectedReply(nil, call.XID, rpc.AuthBadCred)
				if err := ctx.write(reply); err != nil {
					return
				}
				continue
			}
			reply, err := rpc.AppendAcceptedReply(nil, call.XID,
				rpc.Auth{Flavor: rpc.FlavorNone, Body: []byte(rpc.StarttlsToken)},
				rpc.AcceptSuccess, rpc.Mismatch{})
			if err != nil {
				return
			}
			if err := ctx.write(reply); err != nil {
				return
			}
			tlsConn := tls.Server(conn, h.TLSConfig)
			if err := tlsConn.Handshake(); err != nil {
				return
			}
			conn = tlsConn
			ctx.conn = tlsConn
			tlsActive = true
			continue
		}
		// RFC 9289 section 4.1: a server that carries a TLS certificate
		// refuses the procedures of a client that has not upgraded, the
		// NULL procedure of a probe or a ping included only as the probe
		// itself. The plaintext NULL answers, everything else is auth
		// too weak.
		if h.TLSConfig != nil && !tlsActive && call.Procedure != nfs4.ProcNull {
			reply := rpc.AppendRejectedReply(nil, call.XID, rpc.AuthTooWeak)
			if err := ctx.write(reply); err != nil {
				return
			}
			continue
		}
		c := decodeCred(call.Cred)
		if h.RootSquash && c.uid == 0 {
			// Root squash: the claim of uid 0 acts as nobody, the
			// anonymous identity of the export, wherever the credential
			// decides anything afterwards.
			c = cred{uid: nobodyUID, gid: nobodyGID, groups: []uint32{nobodyGID}}
		}
		var reply []byte
		switch {
		case call.Program != nfs4.Program:
			reply, err = rpc.AppendAcceptedReply(nil, call.XID, rpc.AuthNull,
				rpc.AcceptProgUnavail, rpc.Mismatch{})
		case call.Procedure == nfs4.ProcNull && call.Cred.Flavor == rpc.FlavorGSS &&
			len(call.Cred.Body) >= 4 && peekU32(call.Cred.Body) == rpc.GSSVersion3:
			// RPCSEC_GSSv3 control procedures ride on NULLPROC: the
			// verifier and the protected result are generated together
			// under the session lock, in the order a client verifies
			// them.
			body3, verf3, ok3 := h.gssv3Control(call, args)
			if !ok3 {
				reply, err = rpc.AppendAcceptedReply(nil, call.XID, rpc.AuthNull,
					rpc.AcceptGarbageArgs, rpc.Mismatch{})
			} else {
				reply, err = rpc.AppendAcceptedReply(nil, call.XID, verf3,
					rpc.AcceptSuccess, rpc.Mismatch{})
				reply = append(reply, body3...)
			}
		case call.Procedure == nfs4.ProcNull && call.Cred.Flavor == rpc.FlavorGSS &&
			len(call.Cred.Body) >= 4 && peekU32(call.Cred.Body) == rpc.GSSVersion1:
			// RPCSEC_GSS version one control procedures ride the NULL
			// procedure with the context token in the call data, RFC 2203
			// section 5.1.3.
			cred, derr := rpc.DecodeGSSCred(call.Cred.Body)
			switch {
			case derr == nil && cred.Proc == rpc.GSSProcInit:
				body := h.gssInit(args)
				reply, err = rpc.AppendAcceptedReply(nil, call.XID, rpc.AuthNull,
					rpc.AcceptSuccess, rpc.Mismatch{})
				reply = append(reply, body...)
			case derr == nil && cred.Proc == rpc.GSSProcContinue:
				// The krb5 profile establishes a context in one token, so
				// there is nothing to continue with: the initiator sees
				// the major status and starts over.
				body := rpc.AppendGSSInitRes(nil, nil, gssMajorContinueNeeded, 0, 0, nil)
				reply, err = rpc.AppendAcceptedReply(nil, call.XID, rpc.AuthNull,
					rpc.AcceptSuccess, rpc.Mismatch{})
				reply = append(reply, body...)
			case derr == nil && cred.Proc == rpc.GSSProcDestroy:
				if !h.gssDestroy(call) {
					reply = rpc.AppendRejectedReply(nil, call.XID, rpc.AuthGSSCredProb)
				} else {
					reply, err = rpc.AppendAcceptedReply(nil, call.XID, rpc.AuthNull,
						rpc.AcceptSuccess, rpc.Mismatch{})
				}
			default:
				reply, err = rpc.AppendAcceptedReply(nil, call.XID, rpc.AuthNull,
					rpc.AcceptGarbageArgs, rpc.Mismatch{})
			}
		case call.Procedure == nfs4.ProcNull && call.Cred.Flavor == rpc.FlavorGSS:
			// An RPCSEC_GSS credential of a version this server does not
			// speak, sent at the NULL procedure: refuse it as garbage.
			reply, err = rpc.AppendAcceptedReply(nil, call.XID, rpc.AuthNull,
				rpc.AcceptGarbageArgs, rpc.Mismatch{})
		case call.Procedure == nfs4.ProcNull:
			reply, err = rpc.AppendAcceptedReply(nil, call.XID, rpc.AuthNull,
				rpc.AcceptSuccess, rpc.Mismatch{})
		case call.Procedure == nfs4.ProcCompound && call.Cred.Flavor == rpc.FlavorGSS:
			// A COMPOUND under RPCSEC_GSS: the verifier is checked, the
			// arguments unwrapped or verified, the compound runs and the
			// results are protected, all under the session lock, so the
			// tokens leave the server in the order a client verifies
			// them.
			body, verf, ok := h.gssCompound(call, args)
			if !ok {
				reply, err = rpc.AppendAcceptedReply(nil, call.XID, rpc.AuthNull,
					rpc.AcceptGarbageArgs, rpc.Mismatch{})
			} else {
				reply, err = rpc.AppendAcceptedReply(nil, call.XID, verf,
					rpc.AcceptSuccess, rpc.Mismatch{})
				reply = append(reply, body...)
			}
		case call.Procedure == nfs4.ProcCompound:
			body, ok := h.compoundCtx(args, c, ctx)
			if !ok {
				reply, err = rpc.AppendAcceptedReply(nil, call.XID, rpc.AuthNull,
					rpc.AcceptGarbageArgs, rpc.Mismatch{})
			} else {
				reply, err = rpc.AppendAcceptedReply(nil, call.XID, rpc.AuthNull,
					rpc.AcceptSuccess, rpc.Mismatch{})
				reply = append(reply, body...)
			}
		default:
			reply, err = rpc.AppendAcceptedReply(nil, call.XID, rpc.AuthNull,
				rpc.AcceptProcUnavail, rpc.Mismatch{})
		}
		if err != nil {
			return
		}
		if err := ctx.write(reply); err != nil {
			return
		}
	}
}

// decodeCred maps the credential to the identity the operations evaluate
// against. An AUTH_SYS credential carries the client's claim; anything else
// carries nobody.
func decodeCred(a rpc.Auth) cred {
	if a.Flavor == rpc.FlavorSys {
		if sys, err := rpc.DecodeAuthSysBody(a.Body); err == nil {
			return cred{uid: sys.UID, gid: sys.GID, groups: sys.GIDs}
		}
	}
	return cred{uid: 0xffffffff, gid: 0xffffffff}
}

// isSessionSetupOp names the operations a client may run on a fore
// channel without a session; every other operation requires SEQUENCE
// first, per RFC 8881 section 15.1.3.5.
func isSessionSetupOp(op uint32) bool {
	switch op {
	case nfs4.OpExchangeID, nfs4.OpCreateSession, nfs4.OpDestroySession,
		nfs4.OpDestroyClientID, nfs4.OpBindConnToSession, nfs4.OpBackchannelCtl, nfs4.OpSequence:
		return true
	}
	return false
}

// compound executes one COMPOUND4args and returns the COMPOUND4res. A
// second return of false means the arguments were malformed, which is
// GARBAGE_ARGS at the RPC layer and not an NFS status.
//
// A COMPOUND whose first operation is SEQUENCE runs inside a session: the
// slot table of the session drives at-most-once execution, and a repeated
// sequence replays the cached answer. Every other operation of the
// standard requires the session, except the session setup set.
func (h *Handler) compound(args []byte, c cred) ([]byte, bool) {
	return h.compoundCtx(args, c, nil)
}

// compoundCtx is compound with the connection's callback machinery: a
// CREATE_SESSION that negotiates a back channel binds it to this
// connection through ctx.
func (h *Handler) compoundCtx(args []byte, c cred, ctx *connCB) ([]byte, bool) {
	header, d, err := nfs4.DecodeCompoundArgs(args)
	if err != nil {
		return nil, false
	}
	if header.Minor != nfs4.MinorVersion {
		return nfs4.AppendCompoundRes(nil, nfs4.ErrMinorVersMismatch, header.Tag, nil), true
	}

	var reg fhreg
	// The answer accumulates in a single buffer: the header carries
	// placeholders for the top level status and the operation count,
	// patched in place once the operations have run, so no result is
	// copied from an intermediate slice on the way out.
	res := make([]byte, 0, 96+8*int(header.OpCount))
	res = xdr.AppendUint32(res, 0)
	res = xdr.AppendString(res, header.Tag)
	opCountOff := len(res)
	res = xdr.AppendUint32(res, 0)
	nOps := 0
	top := uint32(nfs4.ErrOK)
	var inSession, replayed, cacheThis bool
	var sessID nfs4.SessionID
	var slotID uint32
	var clientID uint64
	for i := range header.OpCount {
		op, err := d.Uint32()
		if err != nil {
			return nil, false
		}
		if op == nfs4.OpSequence && i != 0 {
			top = nfs4.ErrSequencePos
			res = nfs4.AppendOpHeader(res, op, top)
			nOps++
			break
		}
		if op == nfs4.OpSequence {
			a, err := nfs4.DecodeSequenceArgs(d)
			if err != nil {
				return nil, false
			}
			// A lease that lapsed before this request frees the client's
			// state and ends its identity, RFC 8881 section 8.11: the
			// client that comes back after its lease must establish a new
			// one, and nothing of the old life may linger to deny others.
			if h.sessions().leaseExpired(a.SessionID.ClientIDOf(), h.leasePeriod(), time.Now()) {
				h.sessions().destroyClientID(a.SessionID.ClientIDOf())
				h.dropClientState(a.SessionID.ClientIDOf())
				top = nfs4.ErrExpired
				res = nfs4.AppendOpHeader(res, op, top)
				nOps++
				break
			}
			sess, replay, status := h.sessions().sequence(a.SessionID, a.Sequence, a.Slot)
			if status != nfs4.ErrOK {
				top = status
				res = nfs4.AppendOpHeader(res, op, top)
				nOps++
				break
			}
			inSession = true
			sessID, slotID, cacheThis = a.SessionID, a.Slot, a.CacheThis
			clientID = a.SessionID.ClientIDOf()
			reg.clientID = clientID
			h.sessions().renew(clientID, time.Now())
			res = nfs4.AppendSequenceRes(nfs4.AppendOpHeader(res, op, nfs4.ErrOK),
				sessID, a.Sequence, a.Slot, uint32(len(sess.slots)-1), 0)
			nOps++
			if replay {
				_, cached, ok := h.sessions().replay(sessID, slotID)
				if !ok || cached == nil {
					return nfs4.AppendCompoundRes(nil, nfs4.ErrRetryUncachedRep, header.Tag, nil), true
				}
				return cached, true
			}
			continue
		}
		if !inSession && !isSessionSetupOp(op) {
			top = nfs4.ErrOpNotInSession
			res = nfs4.AppendOpHeader(res, op, top)
			nOps++
			break
		}
		if h.sessions().leaseExpired(clientID, h.leasePeriod(), time.Now()) {
			top = nfs4.ErrExpired
			res = nfs4.AppendOpHeader(res, op, top)
			nOps++
			break
		}
		start := time.Now()
		payload, status, err := h.dispatch(op, d, &reg, c, sessID, clientID, ctx)
		if h.LogOps {
			logOp(op, status, time.Since(start))
		}
		if err != nil {
			return nil, false
		}
		res = nfs4.AppendOpHeader(res, op, status)
		nOps++
		// LOCKT answers the DENIED status with the body of the conflicting
		// lock; every other operation carries a body only on success.
		if status == nfs4.ErrOK || (op == nfs4.OpLockt && status == nfs4.ErrDenied) {
			res = append(res, payload...)
		}
		if status != nfs4.ErrOK {
			top = status
			break
		}
	}
	binary.BigEndian.PutUint32(res[0:4], top)
	binary.BigEndian.PutUint32(res[opCountOff:opCountOff+4], uint32(nOps))
	if inSession && !replayed && cacheThis {
		h.sessions().cacheReply(sessID, slotID, top, res)
	}
	return res, true
}

// dispatch executes one operation and returns its result payload, valid
// only while the status is OK, then the status, then an error that marks
// the arguments as malformed rather than the operation as failed.
// opAccess names the permission each data operation requires on the
// current file handle, and on the saved handle where an operation
// crosses the two: the mask the credential must hold before the
// operation touches the backend. Operations outside the table are state
// bookkeeping or handle juggling, whose stateids carry their own
// validation.
func opAccess(op uint32) (mask, savedMask uint32, enforced bool) {
	switch op {
	case nfs4.OpRead, nfs4.OpReadPlus, nfs4.OpSeek, nfs4.OpReadlink,
		nfs4.OpGetxattr, nfs4.OpListxattr, nfs4.OpReaddir:
		return nfsfs.AccessRead, 0, true
	case nfs4.OpLookup, nfs4.OpSecinfo, nfs4.OpSecinfoNoName:
		return nfsfs.AccessLookup, 0, true
	case nfs4.OpWrite, nfs4.OpWriteSame, nfs4.OpSetattr, nfs4.OpCommit,
		nfs4.OpAllocate, nfs4.OpDeallocate, nfs4.OpSetxattr, nfs4.OpRemovexattr:
		return nfsfs.AccessModify, 0, true
	case nfs4.OpCreate, nfs4.OpRemove, nfs4.OpLink:
		return nfsfs.AccessModify, 0, true
	case nfs4.OpRename:
		return nfsfs.AccessModify, nfsfs.AccessModify, true
	case nfs4.OpCopy, nfs4.OpClone:
		return nfsfs.AccessModify, nfsfs.AccessRead, true
	default:
		return 0, 0, false
	}
}

// authorise checks the credential holds the mask on the current handle.
func (h *Handler) authorise(reg *fhreg, mask uint32, c cred) uint32 {
	if !reg.haveCur {
		return nfs4.ErrNoFileHandle
	}
	return h.authoriseHandle(reg.cur, mask, c)
}

// authoriseHandle checks the credential holds the mask on one handle.
func (h *Handler) authoriseHandle(fh nfsfs.Handle, mask uint32, c cred) uint32 {
	granted, err := h.FS.Access(fh, mask, c.uid, c.gid, c.groups)
	if err != nil {
		return mapErr(err)
	}
	if granted&mask != mask {
		return nfs4.ErrAccess
	}
	return nfs4.ErrOK
}

func (h *Handler) dispatch(op uint32, d *xdr.Decoder, reg *fhreg, c cred, sessID nfs4.SessionID, sessionClientid uint64, ctx *connCB) ([]byte, uint32, error) {
	// On a referral stub only the operations that carry the client away
	// answer: everything that touches the backend is MOVED, RFC 5661
	// section 8.4.2.
	if reg.haveCur && h.isReferralStub(reg.cur) &&
		op != nfs4.OpGetattr && op != nfs4.OpGetfh && op != nfs4.OpPutfh &&
		op != nfs4.OpPutRootfh && op != nfs4.OpSavefh && op != nfs4.OpRestorefh &&
		op != nfs4.OpAccess && op != nfs4.OpLookup && op != nfs4.OpSecinfo &&
		op != nfs4.OpSecinfoNoName && op != nfs4.OpSequence {
		if op == nfs4.OpIllegal {
			return nil, nfs4.ErrOpIllegal, nil
		}
		return nil, nfs4.ErrMoved, nil
	}
	// Named attribute handles route to their own operations before the
	// regular dispatch: the synthetic space has no backend behind it.
	if reg.haveCur && h.isNattrFile(reg.cur) {
		switch op {
		case nfs4.OpGetfh:
			return nfs4.AppendGetfhRes(nil, reg.cur), nfs4.ErrOK, nil
		case nfs4.OpGetattr:
			request, rerr := nfs4.ReadBitmap(d)
			if rerr != nil {
				return nil, 0, rerr
			}
			return nfs4.AppendGetattrRes(nil, request, nfs4.Attrs{Type: nfs4.NF4Reg,
				FHExpireType: nfs4.FH4Persistent, NamedAttr: true, UniqueHandles: true}), nfs4.ErrOK, nil
		case nfs4.OpRead:
			if _, rerr := d.Raw(16); rerr != nil {
				return nil, 0, rerr
			}
			off, rerr := d.Uint64()
			if rerr != nil {
				return nil, 0, rerr
			}
			count, rerr := d.Uint32()
			if rerr != nil {
				return nil, 0, rerr
			}
			value, status := h.nattrRead(reg.cur, off)
			if status != nfs4.ErrOK {
				return nil, status, nil
			}
			if uint64(count) < uint64(len(value)) {
				value = value[:count]
			}
			return nfs4.AppendReadRes(nil, true, value), nfs4.ErrOK, nil
		case nfs4.OpWrite:
			if _, rerr := d.Raw(16); rerr != nil {
				return nil, 0, rerr
			}
			off, rerr := d.Uint64()
			if rerr != nil {
				return nil, 0, rerr
			}
			if _, rerr = d.Uint32(); rerr != nil {
				return nil, 0, rerr
			}
			data, rerr := d.VarOpaque()
			if rerr != nil {
				return nil, 0, rerr
			}
			count, status := h.nattrWrite(reg.cur, off, data)
			if status != nfs4.ErrOK {
				return nil, status, nil
			}
			return nfs4.AppendWriteRes(nil, count, nfs4.StableFileSync, h.writeVerifier()), nfs4.ErrOK, nil
		case nfs4.OpSavefh, nfs4.OpRestorefh, nfs4.OpAccess, nfs4.OpCommit:
			return nil, nfs4.ErrOK, nil
		default:
			return nil, nfs4.ErrNotSupp, nil
		}
	}
	if reg.haveCur && h.isNattrDir(reg.cur) {
		switch op {
		case nfs4.OpGetfh:
			return nfs4.AppendGetfhRes(nil, reg.cur), nfs4.ErrOK, nil
		case nfs4.OpGetattr:
			request, rerr := nfs4.ReadBitmap(d)
			if rerr != nil {
				return nil, 0, rerr
			}
			return nfs4.AppendGetattrRes(nil, request, nfs4.Attrs{Type: nfs4.NF4Dir,
				FHExpireType: nfs4.FH4Persistent, NamedAttr: true, UniqueHandles: true}), nfs4.ErrOK, nil
		case nfs4.OpLookup:
			name, rerr := d.String()
			if rerr != nil {
				return nil, 0, rerr
			}
			child, status := h.nattrLookup(reg.cur, name)
			if status != nfs4.ErrOK {
				return nil, status, nil
			}
			reg.cur, reg.haveCur = child, true
			return nil, nfs4.ErrOK, nil
		case nfs4.OpCreate:
			// The named attribute is made with its initial size from the
			// create attributes; the value itself arrives by WRITE. The
			// attributes are a full fattr4 of bitmap and list, RFC 8881
			// section 18.4.
			if _, rerr := d.Uint32(); rerr != nil { // kind, always regular
				return nil, 0, rerr
			}
			name, rerr := d.String()
			if rerr != nil {
				return nil, 0, rerr
			}
			request, rerr := nfs4.ReadBitmap(d)
			if rerr != nil {
				return nil, 0, rerr
			}
			blob, rerr := d.VarOpaque()
			if rerr != nil {
				return nil, 0, rerr
			}
			updates, uerr := nfs4.DecodeSetattrBlob(blob, request)
			if uerr != nil {
				return nil, nfs4.ErrAttrNotSupp, nil
			}
			var value []byte
			if updates.HasSize {
				value = make([]byte, updates.Size)
			}
			if status2 := h.nattrCreate(reg.cur, name, value); status2 != nfs4.ErrOK {
				return nil, status2, nil
			}
			// The CREATE replaces the current handle with the new named
			// attribute file, like every CLAIM_NULL open does.
			child, status2 := h.nattrLookup(reg.cur, name)
			if status2 != nfs4.ErrOK {
				return nil, status2, nil
			}
			reg.cur, reg.haveCur = child, true
			return nfs4.AppendCreateRes(nil), nfs4.ErrOK, nil
		case nfs4.OpRemove:
			name, rerr := d.String()
			if rerr != nil {
				return nil, 0, rerr
			}
			return nfs4.AppendRemoveRes(nil), h.nattrRemove(reg.cur, name), nil
		default:
			return nil, nfs4.ErrNotSupp, nil
		}
	}
	// The permission gate: a data operation must hold the access its
	// mask names on the handle it touches, evaluated against the
	// credential the call carried. State bookkeeping, attribute reads
	// and handle juggling enforce nothing here; their stateids carry
	// their own validation.
	if mask, savedMask, enforced := opAccess(op); enforced {
		if status := h.authorise(reg, mask, c); status != nfs4.ErrOK {
			return nil, status, nil
		}
		if savedMask != 0 {
			if reg.saved == nil {
				return nil, nfs4.ErrNoFileHandle, nil
			}
			if status := h.authoriseHandle(reg.saved, savedMask, c); status != nfs4.ErrOK {
				return nil, status, nil
			}
		}
	}
	switch op {
	case nfs4.OpPutRootfh:
		root, err := h.FS.Root()
		if err != nil {
			return nil, mapErr(err), nil
		}
		reg.cur, reg.haveCur = root, true
		return nil, nfs4.ErrOK, nil

	case nfs4.OpPutfh:
		fh, err := d.VarOpaque()
		if err != nil {
			return nil, 0, err
		}
		reg.cur, reg.haveCur = nfsfs.Handle(fh), true
		return nil, nfs4.ErrOK, nil

	case nfs4.OpSavefh:
		if !reg.haveCur {
			return nil, nfs4.ErrNoFileHandle, nil
		}
		reg.saved = reg.cur
		return nil, nfs4.ErrOK, nil

	case nfs4.OpRestorefh:
		if reg.saved == nil {
			return nil, nfs4.ErrNoFileHandle, nil
		}
		reg.cur, reg.haveCur = reg.saved, true
		return nil, nfs4.ErrOK, nil

	case nfs4.OpGetfh:
		if !reg.haveCur {
			return nil, nfs4.ErrNoFileHandle, nil
		}
		return nfs4.AppendGetfhRes(nil, reg.cur), nfs4.ErrOK, nil

	case nfs4.OpLookup:
		name, err := d.String()
		if err != nil {
			return nil, 0, err
		}
		if !reg.haveCur {
			return nil, nfs4.ErrNoFileHandle, nil
		}
		// A configured referral lands on a stub handle: the client reads
		// the location attributes from it and migrates.
		if r, ok := h.referrals().byName(name); ok {
			reg.cur, reg.haveCur = h.referrals().put(name, r), true
			reg.lastLookup = name
			return nil, nfs4.ErrOK, nil
		}
		child, _, err := h.FS.Lookup(reg.cur, name)
		if err != nil {
			return nil, mapErr(err), nil
		}
		reg.cur, reg.haveCur = child, true
		reg.lastLookup = name
		return nil, nfs4.ErrOK, nil

	case nfs4.OpOpenattr:
		return h.openattrOp(d, reg)

	case nfs4.OpLookupp:
		return h.lookuppOp(reg)

	case nfs4.OpPutPubfh:
		root, err := h.FS.Root()
		if err != nil {
			return nil, mapErr(err), nil
		}
		reg.cur, reg.haveCur = root, true
		return nil, nfs4.ErrOK, nil

	case nfs4.OpVerify:
		return h.verifyOp(d, reg, false)

	case nfs4.OpNverify:
		return h.verifyOp(d, reg, true)

	case nfs4.OpGetattr:
		if !reg.haveCur {
			return nil, nfs4.ErrNoFileHandle, nil
		}
		request, err := nfs4.ReadBitmap(d)
		if err != nil {
			return nil, 0, err
		}
		// A referral stub answers the location attributes instead of a
		// backend lookup, RFC 5661 section 11.9.1.
		if h.isReferralStub(reg.cur) {
			return nfs4.AppendGetattrRes(nil, request, h.stubAttrs(reg.cur)), nfs4.ErrOK, nil
		}
		info, err := h.FS.Getattr(reg.cur)
		if err != nil {
			return nil, mapErr(err), nil
		}
		return nfs4.AppendGetattrRes(nil, request, h.attrsOf(reg.cur, info)), nfs4.ErrOK, nil

	case nfs4.OpAccess:
		if !reg.haveCur {
			return nil, nfs4.ErrNoFileHandle, nil
		}
		mask, err := d.Uint32()
		if err != nil {
			return nil, 0, err
		}
		granted, err := h.FS.Access(reg.cur, mask, c.uid, c.gid, c.groups)
		if err != nil {
			return nil, mapErr(err), nil
		}
		return nfs4.AppendAccessRes(nil, supportedAccess, granted), nfs4.ErrOK, nil

	case nfs4.OpRead:
		if !reg.haveCur {
			return nil, nfs4.ErrNoFileHandle, nil
		}
		var stateid nfs4.Stateid
		raw, err := d.Raw(16)
		if err != nil {
			return nil, 0, err
		}
		copy(stateid[:], raw)
		if status := h.checkStateid(stateid, reg.cur, reg.clientID); status != nfs4.ErrOK {
			return nil, status, nil
		}
		off, err := d.Uint64()
		if err != nil {
			return nil, 0, err
		}
		count, err := d.Uint32()
		if err != nil {
			return nil, 0, err
		}
		if off > 1<<62 {
			return nil, nfs4.ErrInval, nil
		}
		if uint64(count) > nfs4.DefaultLimits.MaxRead {
			count = uint32(nfs4.DefaultLimits.MaxRead)
		}
		// The fast path reads straight into the reply buffer: no
		// intermediate allocation, no second copy, and the end of file
		// comes from the descriptor instead of a second attribute call.
		if ri, ok := h.FS.(nfsfs.ReadIntoer); ok {
			payload := make([]byte, 8, 8+((int(count)+3)&^3))
			n, eof, rerr := ri.ReadInto(reg.cur, int64(off), payload[8:cap(payload)])
			if rerr != nil {
				return nil, mapErr(rerr), nil
			}
			if eof {
				payload[0], payload[1], payload[2], payload[3] = 0, 0, 0, 1
			}
			binary.BigEndian.PutUint32(payload[4:8], uint32(n))
			return payload[:8+((n+3)&^3)], nfs4.ErrOK, nil
		}
		data, err := h.FS.Read(reg.cur, int64(off), int(count))
		if err != nil {
			return nil, mapErr(err), nil
		}
		eof := false
		if info, err := h.FS.Getattr(reg.cur); err == nil {
			eof = int64(off)+int64(len(data)) >= info.Size
		}
		return nfs4.AppendReadRes(nil, eof, data), nfs4.ErrOK, nil

	case nfs4.OpReaddir:
		return h.readdir(d, reg)

	case nfs4.OpWrite:
		return h.writeOp(d, reg)

	case nfs4.OpCreate:
		return h.createOp(d, reg, c)

	case nfs4.OpRemove:
		return h.removeOp(d, reg)

	case nfs4.OpRename:
		return h.renameOp(d, reg)

	case nfs4.OpSetattr:
		return h.setattrOp(d, reg)

	case nfs4.OpLink:
		return h.linkOp(d, reg)

	case nfs4.OpReadlink:
		return h.readlinkOp(d, reg)

	case nfs4.OpCommit:
		return h.commitOp(d, reg)

	case nfs4.OpSecinfo:
		return h.secinfoOp(d, reg)

	case nfs4.OpSecinfoNoName:
		return h.secinfoNoNameOp(d, reg)

	case nfs4.OpExchangeID:
		a, err := nfs4.DecodeExchangeIDArgs(d)
		if err != nil {
			return nil, 0, err
		}
		clientid, sequence, flags, rebooted := h.sessions().exchangeID(a.Verifier, a.OwnerID, time.Now())
		if rebooted != 0 {
			// The old life of this owner is gone: its opens, locks,
			// delegations and layouts go with it, so the rebooted client
			// starts clean and its leftovers deny nobody.
			h.dropClientState(rebooted)
		}
		return nfs4.AppendExchangeIDRes(nil, clientid, sequence, flags, []byte("nfsd")), nfs4.ErrOK, nil

	case nfs4.OpCreateSession:
		a, err := nfs4.DecodeCreateSessionArgs(d)
		if err != nil {
			return nil, 0, err
		}
		id, _, status := h.sessions().createSession(a.ClientID, a.Sequence, a.CBProgram)
		if status != nfs4.ErrOK {
			return nil, status, nil
		}
		// A client that offered a back channel binds it to this
		// connection: the session remembers the callback program and the
		// wire, and CB calls flow through ctx from now on.
		if a.Flags&nfs4.CreateSessionFlagConnBackChan != 0 && ctx != nil {
			h.sessions().attachCB(id, ctx)
		}
		// The answer is deterministic in the values it echoes, so a replay
		// of this sequence regenerates it byte for byte. The flags are
		// echoed as RFC 8881 section 18.36 requires: a client whose
		// CONN_BACK_CHAN offer comes back unanswered tears the client
		// down instead of mounting. The channels are negotiated down to
		// the request: a client rejects a reply larger than what it
		// asked for.
		return nfs4.AppendCreateSessionRes(nil, id, a.Sequence, a.Flags,
			nfs4.NegotiateChannel(a.Fore, nfs4.DefaultForeChannel),
			nfs4.NegotiateChannel(a.Back, nfs4.DefaultBackChannel)), nfs4.ErrOK, nil

	case nfs4.OpDestroySession:
		var id nfs4.SessionID
		raw, err := d.Raw(16)
		if err != nil {
			return nil, 0, err
		}
		copy(id[:], raw)
		// A compound that carries a session may destroy only its own,
		// RFC 8881 section 18.37; the operation travels alone in its
		// compound, so the unguessable session id itself is the
		// credential when no session rode in front.
		if sessionClientid != 0 && id.ClientIDOf() != sessionClientid {
			return nil, nfs4.ErrBadSession, nil
		}
		if status := h.sessions().destroySession(id); status != nfs4.ErrOK {
			return nil, status, nil
		}
		// The layouts and directory delegations of the session die with it.
		h.layouts().dropSession(id)
		h.dirDelegs().dropSession(id)
		return nil, nfs4.ErrOK, nil

	case nfs4.OpOpen:
		return h.openOp(d, reg, c, sessionClientid)

	case nfs4.OpClose:
		return h.closeOp(d, reg, sessionClientid)

	case nfs4.OpLock:
		return h.lockOp(d, reg, c, sessionClientid)

	case nfs4.OpLockt:
		return h.locktOp(d, reg)

	case nfs4.OpLocku:
		return h.lockuOp(d, reg, sessionClientid)

	case nfs4.OpOpenDowngrade:
		return h.openDowngradeOp(d, reg, sessionClientid)

	case nfs4.OpReleaseLockOwner:
		return h.releaseLockOwnerOp(d, sessionClientid)

	case nfs4.OpDelegReturn:
		return h.delegReturnOp(d, sessionClientid)

	case nfs4.OpBackchannelCtl:
		return h.backchannelCtlOp(d, ctx)

	case nfs4.OpBindConnToSession:
		return h.bindConnToSessionOp(d, ctx, sessionClientid)

	case nfs4.OpFreeStateid:
		return h.freeStateidOp(d, sessionClientid)

	case nfs4.OpTestStateid:
		return h.testStateidOp(d)

	case nfs4.OpSeek:
		return h.seekOp(d, reg)

	case nfs4.OpAllocate:
		return h.rangeOp(d, reg, false)

	case nfs4.OpDeallocate:
		return h.rangeOp(d, reg, true)

	case nfs4.OpIoAdvise:
		return h.ioAdviseOp(d, reg)

	case nfs4.OpCopy:
		return h.copyOp(d, reg)

	case nfs4.OpCopyNotify:
		return h.copyNotifyOp(d, reg)

	case nfs4.OpOffloadCancel:
		return h.offloadCancelOp(d, reg)

	case nfs4.OpOffloadStatus:
		return h.offloadStatusOp(d, reg)

	case nfs4.OpClone:
		return h.cloneOp(d, reg)

	case nfs4.OpLayoutError:
		return h.layoutErrorOp(d, reg)

	case nfs4.OpLayoutStats:
		return h.layoutStatsOp(d, reg)

	case nfs4.OpReadPlus:
		return h.readPlusOp(d, reg)

	case nfs4.OpWriteSame:
		return h.writeSameOp(d, reg)

	case nfs4.OpGetxattr:
		return h.getXattrOp(d, reg)

	case nfs4.OpSetxattr:
		return h.setXattrOp(d, reg)

	case nfs4.OpListxattr:
		return h.listXattrOp(d, reg)

	case nfs4.OpRemovexattr:
		return h.removeXattrOp(d, reg)

	case nfs4.OpSetSsv, nfs4.OpWantDelegation:
		return nil, nfs4.ErrNotSupp, nil

	case nfs4.OpGetDirDelegation:
		return h.getDirDelegationOp(d, reg, sessID, sessionClientid)

	case nfs4.OpLayoutGet:
		return h.layoutGetOp(d, reg, sessID, sessionClientid)

	case nfs4.OpLayoutCommit:
		return h.layoutCommitOp(d, reg)

	case nfs4.OpLayoutReturn:
		return h.layoutReturnOp(d, reg, sessID, sessionClientid)

	case nfs4.OpGetDeviceInfo:
		return h.getDeviceInfoOp(d, ctx)

	case nfs4.OpGetDeviceList:
		return h.getDeviceListOp(d, ctx)

	case nfs4.OpDestroyClientID:
		return h.destroyClientIDOp(d, sessionClientid)

	case nfs4.OpReclaimComplete:
		return h.reclaimCompleteOp(d, sessionClientid)

	default:
		if op == nfs4.OpIllegal {
			return nil, nfs4.ErrOpIllegal, nil
		}
		return nil, nfs4.ErrNotSupp, nil
	}
}

// readdir serves one READDIR page: as many entries as the maxcount budget
// takes, in the backend's order, with the verifier and the cookies the
// client resumes from.
func (h *Handler) readdir(d *xdr.Decoder, reg *fhreg) ([]byte, uint32, error) {
	if !reg.haveCur {
		return nil, nfs4.ErrNoFileHandle, nil
	}
	cookie, err := d.Uint64()
	if err != nil {
		return nil, 0, err
	}
	raw, err := d.Raw(8)
	if err != nil {
		return nil, 0, err
	}
	var verifier [8]byte
	copy(verifier[:], raw)
	dircount, err := d.Uint32()
	if err != nil {
		return nil, 0, err
	}
	maxcount, err := d.Uint32()
	if err != nil {
		return nil, 0, err
	}
	request, err := nfs4.ReadBitmap(d)
	if err != nil {
		return nil, 0, err
	}
	if maxcount < 1024 {
		return nil, nfs4.ErrTooSmall, nil
	}

	page, err := h.FS.ReadDir(reg.cur, cookie, 0)
	if err != nil {
		return nil, mapErr(err), nil
	}
	if cookie > 0 && page.Verifier != verifier {
		return nil, nfs4.ErrNotSame, nil
	}

	// The maxcount budget counts every byte of the result body; the
	// dircount budget counts the directory information, approximated here
	// as the cookie and the name of each entry.
	var used, nameUsed int
	var entries []nfs4.DirEntryRes
	for _, e := range page.Entries {
		entry := nfs4.DirEntryRes{Cookie: e.Cookie, Name: e.Name, Attrs: h.attrsOf(e.Handle, e.Info)}
		var buf []byte
		buf = xdr.AppendBool(buf, true)
		buf = xdr.AppendUint64(buf, entry.Cookie)
		buf = xdr.AppendString(buf, entry.Name)
		buf = nfs4.AppendFattr(buf, request, entry.Attrs)
		if used+len(buf)+8 > int(maxcount) ||
			(dircount > 0 && nameUsed+len(e.Name)+8 > int(dircount)) {
			return nfs4.AppendReadDirRes(nil, page.Verifier, entries, request, false), nfs4.ErrOK, nil
		}
		used += len(buf)
		nameUsed += len(e.Name) + 8
		entries = append(entries, entry)
	}
	return nfs4.AppendReadDirRes(nil, page.Verifier, entries, request, true), nfs4.ErrOK, nil
}

// writeOp serves WRITE. The stateid is validated like every stateful
// write; the answer is always FILE_SYNC with the boot verifier, which
// leaves the client nothing to replay after a restart.
func (h *Handler) writeOp(d *xdr.Decoder, reg *fhreg) ([]byte, uint32, error) {
	if !reg.haveCur {
		return nil, nfs4.ErrNoFileHandle, nil
	}
	var stateid nfs4.Stateid
	raw, err := d.Raw(16)
	if err != nil {
		return nil, 0, err
	}
	copy(stateid[:], raw)
	off, err := d.Uint64()
	if err != nil {
		return nil, 0, err
	}
	stable, err := d.Uint32()
	if err != nil {
		return nil, 0, err
	}
	// The data stays in the request record: the write hands the record's
	// own bytes to the backend instead of copying them out first. Raw is
	// bounds checked, so a length beyond the record is refused.
	dataLen, err := d.Uint32()
	if err != nil {
		return nil, 0, err
	}
	dataRaw, err := d.Raw((int(dataLen) + 3) &^ 3)
	if err != nil {
		return nil, 0, err
	}
	data := dataRaw[:dataLen]
	_ = stable
	w := h.writer()
	if w == nil {
		return nil, nfs4.ErrROFS, nil
	}
	// The stateid names the OPEN the write runs under; the anonymous
	// forms are accepted.
	if status := h.checkStateid(stateid, reg.cur, reg.clientID); status != nfs4.ErrOK {
		return nil, status, nil
	}
	if off > 1<<62 {
		return nil, nfs4.ErrInval, nil
	}
	n, err := w.Write(reg.cur, int64(off), data)
	if err != nil {
		return nil, mapErr(err), nil
	}
	return nfs4.AppendWriteRes(nil, uint32(n), nfs4.StableFileSync, h.writeVerifier()), nfs4.ErrOK, nil
}

// createOp serves CREATE, which in NFSv4 makes everything but a regular
// file: directories, symlinks and the special kinds. The exclusive form is
// answered as its guarded equivalent, which matches it for every case but
// a client retrying with the same verifier.
func (h *Handler) createOp(d *xdr.Decoder, reg *fhreg, c cred) ([]byte, uint32, error) {
	w := h.writer()
	if w == nil {
		return nil, nfs4.ErrROFS, nil
	}
	if !reg.haveCur {
		return nil, nfs4.ErrNoFileHandle, nil
	}
	kind, err := d.Uint32()
	if err != nil {
		return nil, 0, err
	}
	var linkdata string
	var major, minor uint32
	switch kind {
	case nfs4.NF4Lnk:
		if linkdata, err = d.String(); err != nil {
			return nil, 0, err
		}
	case nfs4.NF4Blk, nfs4.NF4Chr:
		if major, err = d.Uint32(); err != nil {
			return nil, 0, err
		}
		if minor, err = d.Uint32(); err != nil {
			return nil, 0, err
		}
	}
	name, err := d.String()
	if err != nil {
		return nil, 0, err
	}
	// The object attributes are a full fattr4 of bitmap and list, RFC
	// 8881 section 18.4: CREATE carries no create mode union, that is
	// the OPEN operation's shape.
	request, err := nfs4.ReadBitmap(d)
	if err != nil {
		return nil, 0, err
	}
	blob, err := d.VarOpaque()
	if err != nil {
		return nil, 0, err
	}
	attrs, derr := nfs4.DecodeFattrAttrs(blob, request)
	if derr != nil {
		return nil, 0, derr
	}
	perm := fs.FileMode(attrs.Mode & 0o7777)
	if !request.Has(nfs4.AttrMode) {
		// No mode named: the server default per kind, a writable
		// directory or file.
		if kind == nfs4.NF4Dir {
			perm = 0o755
		} else {
			perm = 0o644
		}
	}
	var spec nfsfs.CreateSpec
	spec = nfsfs.CreateSpec{Kind: kind, Perm: perm}
	spec.LinkData = linkdata
	spec.Major, spec.Minor = major, minor
	// A fresh object carries the owner of the credential that made it,
	// not the daemon's own identity.
	spec.Owner = nfsfs.Owner{UID: c.uid, GID: c.gid}

	h2, _, err := w.Create(reg.cur, name, spec)
	if err != nil {
		return nil, mapErr(err), nil
	}
	dir := reg.cur
	reg.cur, reg.haveCur = h2, true
	h.notifyDirOf(dir, nfs4.OfBits(nfs4.NotifyAddEntry), name)
	return nfs4.AppendCreateRes(nil), nfs4.ErrOK, nil
}

// removeOp serves REMOVE: the named entry of the current directory goes
// away, an empty directory included.
func (h *Handler) removeOp(d *xdr.Decoder, reg *fhreg) ([]byte, uint32, error) {
	w := h.writer()
	if w == nil {
		return nil, nfs4.ErrROFS, nil
	}
	if !reg.haveCur {
		return nil, nfs4.ErrNoFileHandle, nil
	}
	name, err := d.String()
	if err != nil {
		return nil, 0, err
	}
	if err := w.Remove(reg.cur, name); err != nil {
		return nil, mapErr(err), nil
	}
	h.notifyDirOf(reg.cur, nfs4.OfBits(nfs4.NotifyRemoveEntry), name)
	return nfs4.AppendRemoveRes(nil), nfs4.ErrOK, nil
}

// renameOp serves RENAME. The standard fixes the roles of the two file
// handles: the saved one, set by SAVEFH, carries the source directory, and
// the current one carries the target directory.
func (h *Handler) renameOp(d *xdr.Decoder, reg *fhreg) ([]byte, uint32, error) {
	w := h.writer()
	if w == nil {
		return nil, nfs4.ErrROFS, nil
	}
	if !reg.haveCur {
		return nil, nfs4.ErrNoFileHandle, nil
	}
	if reg.saved == nil {
		return nil, nfs4.ErrNoFileHandle, nil
	}
	oldName, err := d.String()
	if err != nil {
		return nil, 0, err
	}
	newName, err := d.String()
	if err != nil {
		return nil, 0, err
	}
	if err := w.Rename(reg.saved, oldName, reg.cur, newName); err != nil {
		return nil, mapErr(err), nil
	}
	// Both directories see the change: the target gains an entry and the
	// source loses one, RFC 8881 section 20.4.
	h.notifyDirOf(reg.cur, nfs4.OfBits(nfs4.NotifyAddEntry), newName)
	h.notifyDirOf(reg.saved, nfs4.OfBits(nfs4.NotifyRemoveEntry), oldName)
	return nfs4.AppendRenameRes(nil), nfs4.ErrOK, nil
}

// setattrOp serves SETATTR: the changes named in the fattr4 are applied to
// the current file. The stateid travels unevaluated: SETATTR is a plain
// backend call. On success the answer names every attribute applied; on
// failure it names none, because the backend applies per call.
func (h *Handler) setattrOp(d *xdr.Decoder, reg *fhreg) ([]byte, uint32, error) {
	w := h.writer()
	if w == nil {
		return nil, nfs4.ErrROFS, nil
	}
	if !reg.haveCur {
		return nil, nfs4.ErrNoFileHandle, nil
	}
	var stateid nfs4.Stateid
	raw, err := d.Raw(16)
	if err != nil {
		return nil, 0, err
	}
	copy(stateid[:], raw)
	request, err := nfs4.ReadBitmap(d)
	if err != nil {
		return nil, 0, err
	}
	blob, err := d.VarOpaque()
	if err != nil {
		return nil, 0, err
	}
	updates, err := nfs4.DecodeSetattrBlob(blob, request)
	if err != nil {
		if errors.Is(err, nfs4.ErrAttrNotSettable) {
			return nil, nfs4.ErrAttrNotSupp, nil
		}
		return nil, 0, err
	}
	if err := w.Setattr(reg.cur, h.setAttrsOf(updates)); err != nil {
		return nil, mapErr(err), nil
	}
	set := nfs4.Bitmap{}
	if updates.HasMode {
		set = set.With(nfs4.AttrMode)
	}
	if updates.HasSize {
		set = set.With(nfs4.AttrSize)
	}
	if updates.UID != nil {
		set = set.With(nfs4.AttrOwner)
	}
	if updates.GID != nil {
		set = set.With(nfs4.AttrOwnerGroup)
	}
	if updates.Atime != nil {
		set = set.With(nfs4.AttrTimeAccessSet)
	}
	if updates.Mtime != nil {
		set = set.With(nfs4.AttrTimeModifySet)
	}
	return nfs4.AppendSetattrRes(nil, set), nfs4.ErrOK, nil
}

// setAttrsOf converts the wire updates into the backend's shape.
func (h *Handler) setAttrsOf(u nfs4.SetAttrUpdates) nfsfs.SetAttrs {
	s := nfsfs.SetAttrs{}
	if u.HasMode {
		mode := u.Mode
		s.Mode = &mode
	}
	if u.HasSize {
		size := int64(u.Size)
		s.Size = &size
	}
	s.UID, s.GID = u.UID, u.GID
	if u.Atime != nil {
		s.Atime = &nfsfs.TimeSet{Now: u.Atime.Server, Time: timeOfNfs(u.Atime.Time)}
	}
	if u.Mtime != nil {
		s.Mtime = &nfsfs.TimeSet{Now: u.Mtime.Server, Time: timeOfNfs(u.Mtime.Time)}
	}
	return s
}

func timeOfNfs(t nfs4.NfsTime) time.Time {
	return time.Unix(t.Seconds, int64(t.Nseconds))
}

// linkOp serves LINK: a hard link named newname lands in the current
// directory and points at the object of the saved file handle.
func (h *Handler) linkOp(d *xdr.Decoder, reg *fhreg) ([]byte, uint32, error) {
	w := h.writer()
	if w == nil {
		return nil, nfs4.ErrROFS, nil
	}
	if !reg.haveCur {
		return nil, nfs4.ErrNoFileHandle, nil
	}
	if reg.saved == nil {
		return nil, nfs4.ErrNoFileHandle, nil
	}
	name, err := d.String()
	if err != nil {
		return nil, 0, err
	}
	if _, _, err := w.Link(reg.saved, reg.cur, name); err != nil {
		return nil, mapErr(err), nil
	}
	return nfs4.AppendLinkRes(nil), nfs4.ErrOK, nil
}

// readlinkOp serves READLINK: the target text of the symlink in the
// current handle. A handle that names anything else is NFS4ERR_INVAL.
func (h *Handler) readlinkOp(d *xdr.Decoder, reg *fhreg) ([]byte, uint32, error) {
	if !reg.haveCur {
		return nil, nfs4.ErrNoFileHandle, nil
	}
	target, err := h.FS.ReadLink(reg.cur)
	if err != nil {
		return nil, mapErr(err), nil
	}
	return nfs4.AppendReadlinkRes(nil, target), nfs4.ErrOK, nil
}

// commitOp serves COMMIT: the backend's dirty data is flushed to stable
// storage and the answer carries the boot verifier, so the client knows
// the flushed writes are the ones it made against this server life.
func (h *Handler) commitOp(d *xdr.Decoder, reg *fhreg) ([]byte, uint32, error) {
	w := h.writer()
	if w == nil {
		return nil, nfs4.ErrROFS, nil
	}
	if !reg.haveCur {
		return nil, nfs4.ErrNoFileHandle, nil
	}
	// The offset, the count and the client's write verifier name the range
	// of the flush; this backend is synchronous, so the whole file is
	// stable whenever COMMIT runs and the range is not evaluated.
	if _, err := d.Uint64(); err != nil {
		return nil, 0, err
	}
	if _, err := d.Uint32(); err != nil {
		return nil, 0, err
	}
	if _, err := d.Raw(8); err != nil {
		return nil, 0, err
	}
	if err := w.Sync(reg.cur); err != nil {
		return nil, mapErr(err), nil
	}
	return nfs4.AppendCommitRes(nil, h.writeVerifier()), nfs4.ErrOK, nil
}

// acceptedSecInfo is the SECINFO answer of this server: the flavours it
// accepts for every name, AUTH_SYS among them. The answer includes
// names that do not exist, because the operation exists precisely so a
// client can probe before it picks its credential.
var acceptedSecInfo = []nfs4.SecinfoEntry{
	{Flavor: nfs4.SecFlavorSys},
}

// secinfoAnswer is shared by both SECINFO operations.
func secinfoAnswer() []byte {
	return nfs4.AppendSecinfoRes(nil, acceptedSecInfo)
}

// secinfoOp serves SECINFO for an explicit name under the current
// directory.
func (h *Handler) secinfoOp(d *xdr.Decoder, reg *fhreg) ([]byte, uint32, error) {
	name, err := d.String()
	if err != nil {
		return nil, 0, err
	}
	if !reg.haveCur {
		return nil, nfs4.ErrNoFileHandle, nil
	}
	if err := nfsfs.ValidName(name); err != nil {
		return nil, mapErr(err), nil
	}
	return secinfoAnswer(), nfs4.ErrOK, nil
}

// secinfoNoNameOp serves SECINFO_NO_NAME. The argument is the
// secinfo_style4 enum alone, RFC 8881 section 18.44: style
// StyleCurrentFH answers for the current file handle, StyleParent for
// its parent directory. The answer carries no name on the wire.
func (h *Handler) secinfoNoNameOp(d *xdr.Decoder, reg *fhreg) ([]byte, uint32, error) {
	style, err := nfs4.DecodeSecinfoNoNameArgs(d)
	if err != nil {
		if errors.Is(err, nfs4.ErrBadStyle) {
			return nil, nfs4.ErrInval, nil
		}
		return nil, 0, err
	}
	if !reg.haveCur {
		return nil, nfs4.ErrNoFileHandle, nil
	}
	switch style {
	case nfs4.StyleCurrentFH, nfs4.StyleParent:
		return secinfoAnswer(), nfs4.ErrOK, nil
	default:
		return nil, nfs4.ErrInval, nil
	}
}

// attrsOf builds the protocol attributes of one file from the backend
// info. The change attribute is the modification time in nanoseconds: the
// finest change counter a local directory offers without extra state, and
// the same clock the access and modify times report.
func (h *Handler) attrsOf(fh nfsfs.Handle, info nfsfs.Info) nfs4.Attrs {
	a := nfs4.Attrs{
		Type:            typeOf(info),
		FHExpireType:    nfs4.FH4Persistent,
		Change:          uint64(info.ModTime.UnixNano()),
		Size:            uint64(info.Size),
		LinkSupport:     true,
		SymlinkSupport:  true,
		NamedAttr:       false,
		FSID:            [2]uint64{info.Dev, 0},
		UniqueHandles:   true,
		FileHandle:      fh,
		FileID:          info.Ino,
		Mode:            uint32(info.Mode.Perm()),
		Numlinks:        uint32(info.Nlink),
		Owner:           strconv.FormatUint(uint64(info.UID), 10),
		OwnerGroup:      strconv.FormatUint(uint64(info.GID), 10),
		SpaceUsed:       uint64(info.Size),
		TimeAccess:      nfsTimeOf(info.ModTime),
		TimeMetadata:    nfsTimeOf(info.ModTime),
		TimeModify:      nfsTimeOf(info.ModTime),
		MountedOnFileID: info.Ino,
		Limits:          nfs4.DefaultLimits,
	}
	return a
}

func nfsTimeOf(t time.Time) nfs4.NfsTime {
	return nfs4.NfsTimeOf(t.Unix(), uint32(t.Nanosecond()))
}

// typeOf maps a Go file mode onto the protocol file type.
func typeOf(info nfsfs.Info) uint32 {
	m := info.Mode
	switch {
	case m&fs.ModeDir != 0:
		return nfs4.NF4Dir
	case m&fs.ModeSymlink != 0:
		return nfs4.NF4Lnk
	case m&fs.ModeDevice != 0 && m&fs.ModeCharDevice != 0:
		return nfs4.NF4Chr
	case m&fs.ModeDevice != 0:
		return nfs4.NF4Blk
	case m&fs.ModeNamedPipe != 0:
		return nfs4.NF4Fifo
	case m&fs.ModeSocket != 0:
		return nfs4.NF4Sock
	default:
		return nfs4.NF4Reg
	}
}

// mapErr turns a backend error into the protocol status it names.
func mapErr(err error) uint32 {
	switch {
	case err == nil:
		return nfs4.ErrOK
	case errors.Is(err, nfsfs.ErrNoEnt):
		return nfs4.ErrNoEnt
	case errors.Is(err, nfsfs.ErrNotDir):
		return nfs4.ErrNotDir
	case errors.Is(err, nfsfs.ErrIsDir):
		return nfs4.ErrIsDir
	case errors.Is(err, nfsfs.ErrStale):
		return nfs4.ErrStale
	case errors.Is(err, nfsfs.ErrNameTooLong):
		return nfs4.ErrNameTooLong
	case errors.Is(err, nfsfs.ErrBadName):
		return nfs4.ErrBadName
	case errors.Is(err, nfsfs.ErrPermission):
		return nfs4.ErrAccess
	case errors.Is(err, nfsfs.ErrReadOnly):
		return nfs4.ErrROFS
	case errors.Is(err, nfsfs.ErrExist):
		return nfs4.ErrExist
	case errors.Is(err, nfsfs.ErrNoSpace):
		return nfs4.ErrNoSpc
	case errors.Is(err, nfsfs.ErrNotEmpty):
		return nfs4.ErrNotEmpty
	case errors.Is(err, nfsfs.ErrInval):
		return nfs4.ErrInval
	case errors.Is(err, nfsfs.ErrNotLnk):
		return nfs4.ErrInval
	case errors.Is(err, nfsfs.ErrIO):
		return nfs4.ErrIO
	default:
		return nfs4.ErrServerFault
	}
}

// probeCount is a test hook: a number that changes between calls, used by
// the tests to build distinct client identities.
func (h *Handler) probeCount() int {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.probes++
	return h.probes
}

// openOp serves OPEN: it opens or creates a regular file under the
// current directory, records the share reservation and hands the client
// the stateid every stateful use of the file will carry. Served claims
// are CLAIM_NULL, CLAIM_PREVIOUS, CLAIM_FH, CLAIM_DELEGATE_CUR and
// CLAIM_DELEGATE_CUR_FH; the exclusive creation forms replay by their
// verifier and answer EXIST on a fresh verifier. A create whose
// attributes name size 0 truncates the existing file.
// exclVerifier answers the create verifier stored for the last
// exclusive create of the name under the directory, if any.
func (h *Handler) exclVerifier(dirKey, name string) ([8]byte, bool) {
	h.exclMu.Lock()
	defer h.exclMu.Unlock()
	if h.excl == nil {
		return [8]byte{}, false
	}
	verf, ok := h.excl[dirKey+"|"+name]
	return verf, ok
}

// setExclVerifier remembers the create verifier of a fresh exclusive
// create, so a client retrying after a lost reply replays into success
// instead of EXIST, RFC 8881 section 18.16. The map is in memory only:
// a restart drops it together with every other live state.
func (h *Handler) setExclVerifier(dirKey, name string, verf [8]byte) {
	h.exclMu.Lock()
	defer h.exclMu.Unlock()
	if h.excl == nil {
		h.excl = make(map[string][8]byte)
	}
	h.excl[dirKey+"|"+name] = verf
}

func (h *Handler) openOp(d *xdr.Decoder, reg *fhreg, c cred, sessionClientid uint64) ([]byte, uint32, error) {
	w := h.writer()
	if w == nil {
		return nil, nfs4.ErrROFS, nil
	}
	if !reg.haveCur {
		return nil, nfs4.ErrNoFileHandle, nil
	}
	a, err := nfs4.DecodeOpenArgs(d)
	if err != nil {
		if errors.Is(err, nfs4.ErrNotSuppName) {
			return nil, nfs4.ErrNotSupp, nil
		}
		return nil, 0, err
	}
	if a.Previous {
		// CLAIM_PREVIOUS: the client reclaims an open it held before the
		// restart. Only valid inside the grace window, before the client
		// announced RECLAIM_COMPLETE, and against a state the store
		// loaded back.
		if ok, gstatus := h.graced().reclaimOKFor(sessionClientid, time.Now()); !ok {
			return nil, gstatus, nil
		}
		if !reg.haveCur {
			return nil, nfs4.ErrNoFileHandle, nil
		}
		stateid, status := h.openStates().reclaimOpen(reg.cur, sessionClientid)
		if status != nfs4.ErrOK {
			return nil, status, nil
		}
		return nfs4.AppendOpenResDeleg(nil, stateid, nfs4.OpenDelegNone, nfs4.Stateid{}), nfs4.ErrOK, nil
	}
	switch a.Claim {
	case nfs4.ClaimDelegatePrev, nfs4.ClaimDelegatePrevFh:
		// The v4.0 delegation reclaim claims name state this server does
		// not track; nothing sane can be answered for them.
		return nil, nfs4.ErrNotSupp, nil
	}
	// Resolve the file the open names. The file handle claims carry no
	// name: the current file handle is the file, nothing is created and
	// RFC 8881 section 18.16 leaves it in place. Only CLAIM_NULL may
	// create; a delegation claim names a file the client already holds.
	// A named open walks the current directory, which the credential
	// must be allowed to search; the file handle claims touch no
	// directory, and the file's own access is checked below.
	if a.Claim == nfs4.ClaimNull || a.Claim == nfs4.ClaimDelegateC {
		if status := h.authoriseHandle(reg.cur, nfsfs.AccessLookup, c); status != nfs4.ErrOK {
			return nil, status, nil
		}
	}
	var fh nfsfs.Handle
	var created bool
	switch a.Claim {
	case nfs4.ClaimFH, nfs4.ClaimDelegateCFh:
		info, gerr := h.FS.Getattr(reg.cur)
		if gerr != nil {
			return nil, mapErr(gerr), nil
		}
		if info.Mode.IsDir() {
			return nil, nfs4.ErrIsDir, nil
		}
		fh = reg.cur
	default:
		create := a.Create && a.Claim == nfs4.ClaimNull
		dirKey := fileKey(reg.cur)
		perm := fs.FileMode(a.Perm & 0o777)
		// An exclusive create without an attribute set names no mode; the
		// file gets the server default rather than no permission bits.
		if a.Exclusive && perm == 0 {
			perm = 0o644
		}
		fh, _, created, err = w.Open(reg.cur, a.Name, create, a.Guarded, a.Truncate, perm,
			nfsfs.Owner{UID: c.uid, GID: c.gid})
		if err != nil {
			// An exclusive create over an existing name answers EXIST,
			// unless the verifier replays the one this server stored for
			// the lost reply of the very create, RFC 8881 section 18.16.
			if a.Exclusive && errors.Is(err, nfsfs.ErrExist) {
				if verf, ok := h.exclVerifier(dirKey, a.Name); ok && verf == a.ExclusiveVerf {
					fh, _, created, err = w.Open(reg.cur, a.Name, false, false, false, perm,
						nfsfs.Owner{UID: c.uid, GID: c.gid})
				}
			}
			if err != nil {
				return nil, mapErr(err), nil
			}
		}
		if a.Exclusive && created {
			h.setExclVerifier(dirKey, a.Name, a.ExclusiveVerf)
		}
	}
	// The opened file must grant the share access the client asked for;
	// a create already carried the directory's modify right through the
	// gate.
	if a.Access&nfs4.ShareAccessRead != 0 {
		if status := h.authoriseHandle(fh, nfsfs.AccessRead, c); status != nfs4.ErrOK {
			return nil, status, nil
		}
	}
	if a.Access&nfs4.ShareAccessWrite != 0 {
		if status := h.authoriseHandle(fh, nfsfs.AccessModify, c); status != nfs4.ErrOK {
			return nil, status, nil
		}
	}
	// A named open replaces the current file handle with the opened file,
	// RFC 8881 section 18.16; a file handle claim already holds it.
	reg.cur, reg.haveCur = fh, true

	// A delegation held by another client conflicts with this open: the
	// recall travels over the holder's back channel on the connection's
	// callback worker, and this open answers NFS4ERR_DELAY while the
	// recall runs, as RFC 8881 section 18.16 prescribes for a conflicting
	// open. The retry after the recall sees the file free.
	key := fileKey(fh)
	if existing, ok := h.delegs().holder(key); ok && existing.clientID != sessionClientid {
		h.queueRecall(existing.sessID, existing.stateid, key, fh)
		return nil, nfs4.ErrDelay, nil
	}

	st, status := h.openStates().open(fh, sessionClientid, a.Owner, a.Access, a.Deny)
	if status != nfs4.ErrOK {
		return nil, status, nil
	}
	h.openStates().persist(h.StateDir)

	// The delegation is granted when this is the file's only open: read
	// only opens carry a read delegation, write opens a write one. The
	// delegation belongs to the session that opened the file, which is
	// where its recalls travel. A delegation claim merges an open into a
	// delegation the client already holds, so no second one is minted.
	// The delegation is granted when this is the file's only open: read
	// only opens carry a read delegation, write opens a write one. The
	// delegation belongs to the session that opened the file, which is
	// where its recalls travel. A delegation claim joins the open to the
	// delegation the client already holds on the file, identified by the
	// client and the file alone, so no second one is minted, RFC 8881
	// section 18.16.4.
	delegType, delegSt := uint32(nfs4.OpenDelegNone), nfs4.Stateid{}
	delegClaim := a.Claim == nfs4.ClaimDelegateC || a.Claim == nfs4.ClaimDelegateCFh
	if h.openStates().countOpens(fh) == 1 && !delegClaim {
		if a.Access == nfs4.ShareAccessRead {
			delegType = nfs4.OpenDelegRead
		} else if a.Access&nfs4.ShareAccessWrite != 0 {
			delegType = nfs4.OpenDelegWrite
		}
		if delegType != nfs4.OpenDelegNone {
			if sessID, ok := h.sessions().sessionOfClient(sessionClientid); ok {
				if delegSt, status = h.delegs().grant(sessID, sessionClientid, key, delegType); status != nfs4.ErrOK {
					delegType = nfs4.OpenDelegNone
				}
			}
		}
		if delegType != nfs4.OpenDelegNone {
			h.openStates().bindDelegation(st, delegSt)
		}
	}
	return nfs4.AppendOpenResDeleg(nil, st, delegType, delegSt), nfs4.ErrOK, nil
}

// closeOp serves CLOSE: the share reservation ends and the stateid the
// answer carries is dead on arrival. An OPEN with byte range locks still
// held is refused with NFS4ERR_LOCKS_HELD.
func (h *Handler) closeOp(d *xdr.Decoder, reg *fhreg, sessionClientid uint64) ([]byte, uint32, error) {
	if !reg.haveCur {
		return nil, nfs4.ErrNoFileHandle, nil
	}
	if h.locks().locksHeldOn(reg.cur) {
		return nil, nfs4.ErrLocksHeld, nil
	}
	// The CLOSE arguments carry the v4.0 seqid ahead of the stateid; it
	// is deprecated and ignored, but it is on the wire, RFC 8881 section
	// 18.2.3.
	if _, err := d.Uint32(); err != nil {
		return nil, 0, err
	}
	var st nfs4.Stateid
	raw, err := d.Raw(16)
	if err != nil {
		return nil, 0, err
	}
	copy(st[:], raw)
	closed, status := h.openStates().close(st, sessionClientid)
	if status != nfs4.ErrOK {
		return nil, status, nil
	}
	h.openStates().persist(h.StateDir)
	// An OPEN whose delegation was granted dies with the open: the last
	// open of the file takes the delegation with it.
	if h.openStates().countOpens(reg.cur) == 0 {
		h.delegs().revoke(fileKey(reg.cur))
	}
	return nfs4.AppendCloseRes(nil, closed), nfs4.ErrOK, nil
}

// checkStateid validates a stateid a stateful operation carries. The
// anonymous forms, all zero and all ones, are accepted without state;
// a real stateid routes by its family mark to the store that minted
// it: OPEN to the open store, DELE to the delegation store, whose
// stateid RFC 8881 section 10.3 lets the client present for its data
// operations. Every stateid must belong to the asking client.
func (h *Handler) checkStateid(st nfs4.Stateid, fh nfsfs.Handle, clientid uint64) uint32 {
	if string(st[4:8]) == "DELE" {
		return h.delegs().checkDataStateid(st, fh, clientid)
	}
	_, status := h.openStates().checkStateid(st, fh, clientid)
	return status
}

// lockOp serves LOCK: a byte range lock hung from an OPEN, either by a new
// lock owner carrying its open stateid, or by an existing lock stateid.
// The locks live beside the share reservations the server tracks; a
// reclaim outside the grace window is refused.
func (h *Handler) lockOp(d *xdr.Decoder, reg *fhreg, c cred, sessionClientid uint64) ([]byte, uint32, error) {
	w := h.writer()
	if w == nil {
		return nil, nfs4.ErrROFS, nil
	}
	if !reg.haveCur {
		return nil, nfs4.ErrNoFileHandle, nil
	}
	lockType, err := d.Uint32()
	if err != nil {
		return nil, 0, err
	}
	if lockType != nfs4.LockTypeRead && lockType != nfs4.LockTypeWrite {
		return nil, nfs4.ErrInval, nil
	}
	reclaim, err := d.Bool()
	if err != nil {
		return nil, 0, err
	}
	offset, err := d.Uint64()
	if err != nil {
		return nil, 0, err
	}
	length, err := d.Uint64()
	if err != nil {
		return nil, 0, err
	}
	newOwner, err := d.Bool()
	if err != nil {
		return nil, 0, err
	}
	var lockClientid uint64
	var lockOwner []byte
	if newOwner {
		// The locker union carries the open stateid the lock hangs from
		// and the identity of the lock owner: open_seqid, open_stateid,
		// lock_seqid, clientid, owner. The open stateid must name a live
		// open of this file that belongs to the asking client, and the
		// lock is registered under the session's client, never under a
		// clientid the wire alone claims, RFC 8881 section 18.10.
		if _, err = d.Uint32(); err != nil {
			return nil, 0, err
		}
		raw, rerr := d.Raw(16)
		if rerr != nil {
			return nil, 0, rerr
		}
		var lockSt nfs4.Stateid
		copy(lockSt[:], raw)
		if _, err = d.Uint32(); err != nil {
			return nil, 0, err
		}
		if _, err = d.Uint64(); err != nil { // locker4 open owner: clientid
			return nil, 0, err
		}
		if lockOwner, err = d.VarOpaque(); err != nil {
			return nil, 0, err
		}
		if _, status := h.openStates().checkStateid(lockSt, reg.cur, sessionClientid); status != nfs4.ErrOK {
			return nil, nfs4.ErrBadStateid, nil
		}
		lockClientid = sessionClientid
	} else {
		// An existing lock owner: the stateid names the lock state.
		raw, err := d.Raw(16)
		if err != nil {
			return nil, 0, err
		}
		var lockSt nfs4.Stateid
		copy(lockSt[:], raw)
		if _, err = d.Uint32(); err != nil {
			return nil, 0, err
		}
		ls, status := h.locks().byStateid(lockSt, reg.cur, sessionClientid)
		if status != nfs4.ErrOK {
			return nil, status, nil
		}
		lockClientid, lockOwner = ls.clientID, ls.owner
	}

	// A reclaim re-establishes a lock held before the restart. The server
	// recovers no lock state itself, so inside the grace window the
	// reclaim re-registers the lock from the presented identity; after the
	// window it is NO_GRACE (RFC 8881 section 13.12).
	if reclaim && !h.graced().active(time.Now()) {
		return nil, nfs4.ErrNoGrace, nil
	}

	st, status := h.locks().lock(reg.cur, lockClientid, lockOwner,
		lockType == nfs4.LockTypeWrite, offset, length)
	if status == nfs4.ErrDenied {
		// A denied lock is remembered, so the release of the conflicting
		// range can notify this owner over its back channel. The routing
		// identity is the session client, the wire owner names the lock
		// owner within it.
		h.locks().addWaiter(reg.cur, sessionClientid, lockOwner, offset, length,
			lockType == nfs4.LockTypeWrite)
		return nil, status, nil
	}
	if status != nfs4.ErrOK {
		return nil, status, nil
	}
	return nfs4.AppendLockRes(nil, st), nfs4.ErrOK, nil
}

// locktOp serves LOCKT serves LOCKT serves LOCKT: a probe whether a lock over the range would
// conflict with another owner's locks on the current file. A conflict is
// answered NFS4ERR_DENIED with the holder of the lock.
func (h *Handler) locktOp(d *xdr.Decoder, reg *fhreg) ([]byte, uint32, error) {
	if !reg.haveCur {
		return nil, nfs4.ErrNoFileHandle, nil
	}
	lockType, err := d.Uint32()
	if err != nil {
		return nil, 0, err
	}
	if lockType != nfs4.LockTypeRead && lockType != nfs4.LockTypeWrite {
		return nil, nfs4.ErrInval, nil
	}
	offset, err := d.Uint64()
	if err != nil {
		return nil, 0, err
	}
	length, err := d.Uint64()
	if err != nil {
		return nil, 0, err
	}
	clientid, err := d.Uint64()
	if err != nil {
		return nil, 0, err
	}
	owner, err := d.VarOpaque()
	if err != nil {
		return nil, 0, err
	}
	denied, status := h.locks().test(reg.cur, clientid, owner,
		lockType == nfs4.LockTypeWrite, offset, length)
	if status == nfs4.ErrDenied {
		return nfs4.AppendLocktResDenied(nil, denied.Offset, denied.Length,
			denied.LockType, denied.ClientID, denied.Owner), nfs4.ErrDenied, nil
	}
	if status != nfs4.ErrOK {
		return nil, status, nil
	}
	return nfs4.AppendLocktResOK(nil), nfs4.ErrOK, nil
}

// lockuOp serves LOCKU: the release of one range of a lock stateid. The
// answer carries the stateid with a bumped sequence.
func (h *Handler) lockuOp(d *xdr.Decoder, reg *fhreg, sessionClientid uint64) ([]byte, uint32, error) {
	if !reg.haveCur {
		return nil, nfs4.ErrNoFileHandle, nil
	}
	if _, err := d.Uint32(); err != nil { // lock type, echoed
		return nil, 0, err
	}
	if _, err := d.Uint32(); err != nil { // seqid, deprecated
		return nil, 0, err
	}
	var st nfs4.Stateid
	raw, err := d.Raw(16)
	if err != nil {
		return nil, 0, err
	}
	copy(st[:], raw)
	offset, err := d.Uint64()
	if err != nil {
		return nil, 0, err
	}
	length, err := d.Uint64()
	if err != nil {
		return nil, 0, err
	}
	closed, status := h.locks().unlock(st, sessionClientid, offset, length)
	if status != nfs4.ErrOK {
		return nil, status, nil
	}
	// Waiters whose conflict the release may have lifted learn about it
	// over their back channel; the notification is advisory.
	for _, w := range h.locks().takeWaiters(reg.cur, offset, length) {
		h.notifyLockAvailable(reg.cur, w.clientID, w.owner)
	}
	return nfs4.AppendLockuRes(nil, closed), nfs4.ErrOK, nil
}

// notifyLockAvailable pushes CB_NOTIFY_LOCK to the session of the lock
// owner a denied lock was recorded for. The notification is queued onto
// the connection's callback worker, so a LOCKU served on the waiter's
// own connection never waits for the reply only that connection's read
// loop can route.
func (h *Handler) notifyLockAvailable(fh nfsfs.Handle, clientid uint64, owner []byte) {
	sessID, ok := h.sessions().sessionOfClient(clientid)
	if !ok {
		return
	}
	args := nfs4.AppendCBNotifyLockArgs(nil, fh, clientid, owner)
	_ = h.queueCB(sessID, "lock-notify", [][]byte{args})
}

// openDowngradeOp serves OPEN_DOWNGRADE: the share access and deny bits of
// a live OPEN narrow down and the stateid sequence moves one up.
func (h *Handler) openDowngradeOp(d *xdr.Decoder, reg *fhreg, sessionClientid uint64) ([]byte, uint32, error) {
	if !reg.haveCur {
		return nil, nfs4.ErrNoFileHandle, nil
	}
	// The deprecated v4.0 seqid rides ahead of the stateid, RFC 8881
	// section 18.7.3.
	if _, err := d.Uint32(); err != nil {
		return nil, 0, err
	}
	var st nfs4.Stateid
	raw, err := d.Raw(16)
	if err != nil {
		return nil, 0, err
	}
	copy(st[:], raw)
	access, err := d.Uint32()
	if err != nil {
		return nil, 0, err
	}
	deny, err := d.Uint32()
	if err != nil {
		return nil, 0, err
	}
	closed, status := h.openStates().downgrade(st, sessionClientid, access, deny)
	if status != nfs4.ErrOK {
		return nil, status, nil
	}
	_ = closed
	return nfs4.AppendOpenDowngradeRes(nil), nfs4.ErrOK, nil
}

// destroyClientIDOp serves DESTROY_CLIENTID: the client, its sessions
// and its OPEN state go away. A compound that carries a session may
// destroy only its own client, RFC 8881 section 18.50; one without a
// session authenticates by holding the unguessable client id. A client
// id the server does not know is NFS4ERR_STALE_CLIENTID.
func (h *Handler) destroyClientIDOp(d *xdr.Decoder, sessionClientid uint64) ([]byte, uint32, error) {
	clientid, err := d.Uint64()
	if err != nil {
		return nil, 0, err
	}
	if sessionClientid != 0 && clientid != sessionClientid {
		return nil, nfs4.ErrClientIDBusy, nil
	}
	if !h.sessions().destroyClientID(clientid) {
		return nil, nfs4.ErrStaleClientID, nil
	}
	h.dropClientState(clientid)
	return nil, nfs4.ErrOK, nil
}

// reclaimCompleteOp serves RECLAIM_COMPLETE: the client announces it has
// reclaimed everything it could. A second announcement is
// NFS4ERR_COMPLETE_ALREADY, and an announcement after the grace window
// closed is NFS4ERR_NO_GRACE.
func (h *Handler) reclaimCompleteOp(d *xdr.Decoder, sessionClientid uint64) ([]byte, uint32, error) {
	oneClient, err := d.Bool()
	if err != nil {
		return nil, 0, err
	}
	clientid := sessionClientid
	if oneClient {
		// The rca_one_client form names the client on behalf of another;
		// this build answers it for the caller alone, which is the only
		// client it may speak for.
		if _, err = d.Uint64(); err != nil {
			return nil, 0, err
		}
	}
	if !h.sessions().knownClient(clientid) {
		return nil, nfs4.ErrStaleClientID, nil
	}
	if !h.graced().active(time.Now()) {
		return nil, nfs4.ErrNoGrace, nil
	}
	if !h.graced().complete(clientid, time.Now()) {
		return nil, nfs4.ErrCompleteAlready, nil
	}
	return nil, nfs4.ErrOK, nil
}

// SendCB delivers one CB_COMPOUND to the session's back channel on the
// connection's callback worker and waits for the reply. It serves
// callers on their own goroutines; the request path uses queueRecall
// and queueCB, which never block the dispatching connection.
func (h *Handler) SendCB(sessID nfs4.SessionID, tag string, ops [][]byte) (nfs4.CompoundRes, [][]byte, error) {
	return h.sessions().callCB(sessID, tag, ops)
}

// queueCB posts one fire and forget CB_COMPOUND to the session's back
// channel.
func (h *Handler) queueCB(sessID nfs4.SessionID, tag string, ops [][]byte) error {
	return h.sessions().queueCB(sessID, tag, ops, nil)
}

// queueRecall recalls one delegation over the holder's back channel
// without blocking the caller: the recall, and the revocation that
// follows it whatever the delivery outcome, run on the holder's
// connection worker. A recall that cannot even be queued, because the
// session or its back channel is gone, revokes at once; the revocation
// lands only on the very delegation the recall named, so a grant that
// arrived in the meantime survives.
func (h *Handler) queueRecall(sessID nfs4.SessionID, st nfs4.Stateid, key string, fh nfsfs.Handle) {
	recall := [][]byte{nfs4.AppendCBRecallArgs(nil, st, false, fh)}
	err := h.sessions().queueCB(sessID, "recall", recall, func(cbResult) {
		h.delegs().revokeIf(key, st)
	})
	if err != nil {
		h.delegs().revokeIf(key, st)
	}
}

// dropClientState releases every piece of state a client holds: its
// opens, locks, delegations, layouts and directory delegations. It is
// what DESTROY_CLIENTID, a rebooting EXCHANGE_ID and a lapsed lease
// require, so none of them leaves state behind under an identity that
// never comes back.
func (h *Handler) dropClientState(clientid uint64) {
	h.openStates().dropClient(clientid)
	h.locks().dropClient(clientid)
	h.delegs().dropClient(clientid)
	h.layouts().dropClient(clientid)
	h.dirDelegs().dropClient(clientid)
}