mesh-delivery/modules/mesh-delivery-checker/internal/check/requests.go

package check

import (
	"bytes"
	"compress/gzip"
	"context"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"sort"
	"strings"
	"time"
)

// The composition requests a check sends (issue 1921, notes 66-67): the request the checker sends the composer for one
// node, the checking program's whole input of a composition, is kept in the checker's own state RequestsStateName, so a
// session reproduces a check's composition exactly. The checker is a foundation part (novox/hq ADR 0619 §1-3), so its
// kept data is a section of its manifest's old form, here `state`, and not an object store asked in `requires`.
//
// **What bounds it.** A request is the node-engine's whole composition input of one node, about 430 KB as JSON. A
// state's value holds at most 256 KiB (kvlimit.ModuleStateValueBytes), so it is kept gzip-compressed (JSON shrinks about
// tenfold) in a document of its own; a request whose compressed bytes still do not fit is refused by the put, which
// fails the node's composition naming the size (never dropped). The manifest's ttl-seconds is its lifetime (3 days):
// the bus drops a value past it, so nothing waits on a timer here. A check kept again overwrites the key.
const (
	// RequestsStateName is the checker's state of kept requests and RequestsBucket its bucket.
	RequestsStateName = "compositions"
	RequestsBucket    = CheckerModule + "_" + RequestsStateName
	// RequestsReadTool is the tool that reads a kept request.
	RequestsReadTool = "checker_composition_requests"
)

// KeptRequest is a request as the state holds it: whose it is, and the request's JSON, compressed.
type KeptRequest struct {
	V      int       `json:"v"`
	Check  string    `json:"check"`
	Node   string    `json:"node"`
	KeptAt time.Time `json:"kept_at"`
	// Size and Digest are the request's, uncompressed: its bytes and their sha256.
	Size   int    `json:"size"`
	Digest string `json:"digest"`
	// Gzip is the request's bytes, gzip-compressed and base64-encoded.
	Gzip string `json:"gzip"`
}

// RequestKey is the key a check's request to a node is kept under: the check's key (group.heads) and the node.
func RequestKey(key, node string) string { return key + "/" + node }

// RequestPutter writes a value of the state, replacing one of that key.
type RequestPutter interface {
	Put(ctx context.Context, key string, value []byte) (uint64, error)
}

// RequestKeeper keeps the requests of a check.
type RequestKeeper struct {
	Store RequestPutter
	Now   func() time.Time
}

// Keep stores body under the check's key and the node. An error says why it could not, for the verdict's words.
func (k RequestKeeper) Keep(ctx context.Context, key, node string, body []byte) error {
	var packed bytes.Buffer
	zw := gzip.NewWriter(&packed)
	if _, err := zw.Write(body); err != nil {
		return fmt.Errorf("the composition request of check %s to %s could not be compressed: %w", key, node, err)
	}
	if err := zw.Close(); err != nil {
		return fmt.Errorf("the composition request of check %s to %s could not be compressed: %w", key, node, err)
	}
	now := time.Now
	if k.Now != nil {
		now = k.Now
	}
	sum := sha256.Sum256(body)
	doc, err := json.Marshal(KeptRequest{V: 1, Check: key, Node: node, KeptAt: now().UTC(), Size: len(body),
		Digest: hex.EncodeToString(sum[:]), Gzip: base64.StdEncoding.EncodeToString(packed.Bytes())})
	if err != nil {
		return err
	}
	if _, err := k.Store.Put(ctx, RequestKey(key, node), doc); err != nil {
		return fmt.Errorf("the composition request of check %s to %s (%d bytes, %d compressed) could not be kept in the state %s: %w",
			key, node, len(body), packed.Len(), RequestsStateName, err)
	}
	return nil
}

// ReadRequests answers the kept requests: with a check and a node, that request whole; with a check, the nodes it kept
// one for; with neither, every kept request. A node without a check is refused. get is the state's value of a key.
func ReadRequests(s StateRead, key, node string) (any, error) {
	if s.Keys == nil || s.Get == nil {
		return nil, fmt.Errorf("the state %s cannot be read here", RequestsStateName)
	}
	if node != "" && key == "" {
		return nil, fmt.Errorf("a node is asked for with the check's key (group.heads) of the request")
	}
	if node != "" {
		raw, err := s.Get(RequestKey(key, node))
		if err != nil {
			return nil, err
		}
		if len(raw) == 0 {
			return nil, fmt.Errorf("no composition request of check %s to %s is kept (none was sent, or it is past its lifetime)", key, node)
		}
		kept, body, err := unpackRequest(raw)
		if err != nil {
			return nil, fmt.Errorf("the kept request %s: %w", RequestKey(key, node), err)
		}
		return map[string]any{"check": kept.Check, "node": kept.Node, "size": kept.Size, "digest": kept.Digest, "kept_at": kept.KeptAt, "request": json.RawMessage(body)}, nil
	}
	keys, err := s.Keys()
	if err != nil {
		return nil, err
	}
	sort.Strings(keys)
	listed := []map[string]any{}
	for _, k := range keys {
		if key != "" && !strings.HasPrefix(k, key+"/") {
			continue
		}
		raw, err := s.Get(k)
		if err != nil {
			return nil, err
		}
		var kept KeptRequest
		if err := json.Unmarshal(raw, &kept); err != nil {
			listed = append(listed, map[string]any{"key": k, "unreadable": err.Error()})
			continue
		}
		listed = append(listed, map[string]any{"check": kept.Check, "node": kept.Node, "size": kept.Size, "digest": kept.Digest, "kept_at": kept.KeptAt})
	}
	return map[string]any{"kept": listed}, nil
}

// unpackRequest reads a kept document and its request's bytes, which must match the digest they were kept with.
func unpackRequest(raw []byte) (KeptRequest, []byte, error) {
	var kept KeptRequest
	if err := json.Unmarshal(raw, &kept); err != nil {
		return kept, nil, fmt.Errorf("is no kept request: %w", err)
	}
	packed, err := base64.StdEncoding.DecodeString(kept.Gzip)
	if err != nil {
		return kept, nil, fmt.Errorf("holds no base64: %w", err)
	}
	zr, err := gzip.NewReader(bytes.NewReader(packed))
	if err != nil {
		return kept, nil, fmt.Errorf("holds no gzip: %w", err)
	}
	body, err := io.ReadAll(zr)
	if err != nil {
		return kept, nil, fmt.Errorf("holds no complete gzip: %w", err)
	}
	if sum := sha256.Sum256(body); hex.EncodeToString(sum[:]) != kept.Digest {
		return kept, nil, fmt.Errorf("does not match the digest it was kept with")
	}
	return kept, body, nil
}