mesh-catalog/modules/mounts/cmd/mounts/bus.go

package main

// The bundle's half on the bus (novox/hq ADR 0263 rule 9, ADR 0201). The bundle runs as the operator
// account and reaches the bus through the runtime; the process runs as root and does not. So every 30
// seconds the bundle:
//
//   - reads what the sharing machine exports from nfs-server's state `exports`, keyed by that machine
//     (ADR 0201: a module reads another's state, never its files), and leaves it in ExportsFile for the
//     process. A failed read keeps the last answer's shares, so a bus that is away for a minute removes no
//     mount;
//   - puts what the process last found in the state `shares`, keyed by this machine, so a view or the
//     controller watches it without reading this machine's files.

import (
	"encoding/json"
	"fmt"
	"git.novox.be/novox/mesh-sdk/go/health"
	"os"
	"path/filepath"
	"time"
)

// SharesState is the state the module keeps on the bus, one key per machine.
const SharesState = "shares"

// ExportsState is nfs-server's state this module reads: what each sharing machine exports, by machine.
const ExportsState = "nfs-server.exports"

// Getter reads one key of a state, nil when it has none; Putter puts one key of a state.
type (
	Getter func(key string) (json.RawMessage, error)
	Putter func(key string, value any) error
)

// AskExports reads what the sharing machine exports, once, and answers what to leave for the process.
func AskExports(cfg Config, previous ExportsAnswer, get Getter, now time.Time) ExportsAnswer {
	out := ExportsAnswer{Server: cfg.ServerNode(), At: cfg.At, Asked: now.Format(time.RFC3339)}
	if out.Server == "" {
		out.Error = "no machine shares nfs-share for this one"
		return out
	}
	raw, err := get(out.Server)
	var answer struct {
		Shares []ServerShare `json:"shares"`
	}
	if err == nil && len(raw) == 0 {
		err = fmt.Errorf("%s keeps nothing in nfs-server's state exports yet", out.Server)
	}
	if err == nil {
		err = json.Unmarshal(raw, &answer)
		if err == nil && answer.Shares == nil {
			err = fmt.Errorf("the answer names no shares")
		}
	}
	if err != nil {
		out.Error = err.Error()
		if previous.Server == out.Server {
			out.Shares, out.Answered = previous.Shares, previous.Answered
		}
		return out
	}
	out.Shares, out.Answered = answer.Shares, out.Asked
	return out
}

// SharesValue is what the module keeps on the bus for one machine: what its process last found, and what
// the machine asks of which server, so that the server exports a share to the machines that ask for it
// and to nobody else (the review of 2026-10-08, finding 5). The server decides the access it gives.
type SharesValue struct {
	Status
	Wants []Want `json:"wants"`
}

// Share is one pass of the bundle's work over a machine.
func (m *Machine) Share(node string, get Getter, put Putter, last *string, now time.Time) []string {
	var said []string
	wants := []Want{}
	if text, err := m.read(ConfigFile); err == nil {
		cfg := ParseConfig(text)
		wants = cfg.Wants()
		if len(cfg.Shares) > 0 {
			previous, _ := readJSON[ExportsAnswer](m, ExportsFile)
			answer := AskExports(cfg, previous, get, now)
			if answer.Error != "" {
				said = append(said, "what the sharing machine exports was not read: "+answer.Error)
			}
			if err := m.leave(ExportsFile, answer); err != nil {
				said = append(said, "the answer was not left for the process: "+err.Error())
			}
		}
	}
	if node != "" && put != nil {
		value := SharesValue{Wants: wants}
		if text, err := m.read(StatusFile); err == nil {
			json.Unmarshal([]byte(text), &value.Status)
		}
		if value.Status.Shares == nil {
			value.Status.Shares = []ShareStatus{}
		}
		raw, _ := json.Marshal(value)
		if string(raw) != *last {
			if err := put(node, value); err != nil {
				said = append(said, "the state was not put: "+err.Error())
			} else {
				*last = string(raw)
			}
		}
	}
	return said
}

// leave writes a file of the bundle's for the process, through a temporary beside it.
func (m *Machine) leave(p string, v any) error {
	raw, _ := json.MarshalIndent(v, "", "  ")
	full := filepath.Join(m.Root, p)
	tmp := full + ".new"
	if err := os.WriteFile(tmp, append(raw, '\n'), 0o644); err != nil {
		return err
	}
	return os.Rename(tmp, full)
}

// Asker calls a tool on the mesh as this module.
type Asker func(key string, body any) (json.RawMessage, error)

// Readings reads disk-load's readings of one machine's data items (its state `readings`, ADR 0571 clause 3).
type Readings func(node string) ([]health.Reading, error)

// LeaveData reads disk-load for where the mesh keeps data on this machine and leaves the paths for the
// process (finding 4d). A failed ask keeps the last paths, with why; with none, the process sets up
// nothing new.
func (m *Machine) LeaveData(node string, read Readings, now time.Time) []string {
	if node == "" || read == nil {
		return nil
	}
	previous, _ := readJSON[DataPaths](m, DataFile)
	out := DataPaths{Machine: node, Asked: now.Format(time.RFC3339)}
	readings, err := read(node)
	var paths []string
	if err == nil {
		paths, err = PathsFrom(readings)
	}
	var said []string
	if err != nil {
		out.Error = err.Error()
		if previous.Machine == node {
			out.Paths, out.Answered = previous.Paths, previous.Answered
		}
		said = append(said, "where the mesh keeps data here was not read: "+err.Error())
	} else {
		out.Paths, out.Answered = paths, out.Asked
	}
	if err := m.leave(DataFile, out); err != nil {
		said = append(said, "the data list was not left for the process: "+err.Error())
	}
	return said
}