mesh-tools/node-tools/internal/bus/pages.go

package bus

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/nats-io/nats.go"

	"github.com/novox/mesh-tools/node-tools/internal/wire"
)

// An answer larger than the bus carries in one message (novox/hq issue 314).
//
// The bus carries one message of at most its max_payload, and the client library refuses a larger one
// before it leaves the process that sent it. Until this, a holder whose answer was too large wrote one
// line to its own log and its caller waited out RequestTimeout and read "did not answer" — the
// controller's `conditions` and `status` for hours on 2026-10-08, and the operator's channel with them.
//
// The controller pages such an answer (mesh-controller internal/link, pages.go): it answers an error
// carrying `paged` — the call it holds the answer under, its size and its parts — and a caller asks the
// same subject again once per part with PageHeader, each part answered raw with PartHeader. Asking the
// same subject means no grant beyond the one the caller already had. A module's answer too large is
// refused here, loudly, in a message that fits: its instances may be several, and a part asked again
// could reach another.

const (
	// PageHeader asks for one part of a paged answer: `<call> <part>`, from zero.
	PageHeader = "Mesh-Page"
	// PartHeader is on a part: `<part>/<parts>`.
	PartHeader = "Mesh-Part"
	// partTries is how often one part is asked: during a handover a part can reach the controller that
	// did not hold it.
	partTries = 3
)

// PartTimeout bounds the asking of one part.
var PartTimeout = 10 * time.Second

// paged is what a too-large answer says instead of itself.
type paged struct {
	Call  string `json:"call"`
	Bytes int    `json:"bytes"`
	Parts int    `json:"parts"`
}

func pagedIn(data []byte) (paged, bool) {
	if !bytes.Contains(data, []byte(`"paged"`)) {
		return paged{}, false
	}
	var r struct {
		Paged *paged `json:"paged"`
	}
	if json.Unmarshal(data, &r) != nil || r.Paged == nil || r.Paged.Call == "" || r.Paged.Parts < 1 {
		return paged{}, false
	}
	return *r.Paged, true
}

// whole is a reply's whole answer: the reply itself, or every part of a paged one, joined. A part that
// cannot be had is an error naming it, never a shorter answer.
func (c *Conn) whole(subject string, data []byte) ([]byte, error) {
	p, ok := pagedIn(data)
	if !ok {
		return data, nil
	}
	out := make([]byte, 0, p.Bytes)
	for n := 0; n < p.Parts; n++ {
		var part []byte
		var err error
		for try := 0; try < partTries; try++ {
			if part, err = c.askPart(subject, p, n); err == nil {
				break
			}
		}
		if err != nil {
			return nil, fmt.Errorf("the answer was %d bytes, more than the bus carries in one message, paged as %s "+
				"in %d parts, and part %d could not be had: %w", p.Bytes, p.Call, p.Parts, n, err)
		}
		out = append(out, part...)
	}
	if len(out) != p.Bytes {
		return nil, fmt.Errorf("the answer paged as %s was %d bytes, and its parts joined are %d", p.Call, p.Bytes, len(out))
	}
	return out, nil
}

func (c *Conn) askPart(subject string, p paged, n int) ([]byte, error) {
	ask := nats.NewMsg(subject)
	ask.Header.Set(PageHeader, fmt.Sprintf("%s %d", p.Call, n))
	ask.Data = []byte(`{}`)
	reply, err := c.nc.RequestMsg(ask, PartTimeout)
	if err != nil {
		return nil, err
	}
	if got := reply.Header.Get(PartHeader); got != fmt.Sprintf("%d/%d", n, p.Parts) {
		var r struct {
			Error string `json:"error"`
		}
		if json.Unmarshal(reply.Data, &r) == nil && r.Error != "" {
			return nil, errors.New(r.Error)
		}
		return nil, fmt.Errorf("asked part %d/%d, answered %q", n, p.Parts, got)
	}
	return reply.Data, nil
}

// fits is an answer the bus can carry: the answer itself, or — when it is larger than one message —
// an error that says so, which is.
func (c *Conn) fits(subject string, body []byte) []byte {
	limit := c.nc.MaxPayload()
	if limit <= 0 || int64(len(body)) <= limit {
		return body
	}
	c.Logf("[mesh-tools] the answer on %s is %d bytes, more than the bus carries in one message (%d): its caller "+
		"is told so instead", subject, len(body), limit)
	said, _ := wire.Marshal(reply{Node: c.node, Error: fmt.Sprintf("the answer is %d bytes, more than the bus "+
		"carries in one message (%d), so it was not sent: ask for less — a narrower question, a shorter window, "+
		"fewer lines", len(body), limit)})
	return said
}