diff --git a/modules/disk-load/README.md b/modules/disk-load/README.md index 4ba20e5e..4d4f764e 100644 --- a/modules/disk-load/README.md +++ b/modules/disk-load/README.md @@ -3,7 +3,9 @@ How loaded a machine's storage is, read from `/proc` and `/sys` and nothing else: the share of CPU time spent waiting on I/O, I/O pressure stall, each disk's busy share, throughput, IOPS, wait and queue depth, the filesystems' space, and the units, containers and processes doing the I/O (novox/hq issue -312). It changes nothing and declares no resources, so it can be assigned to every machine. +312). It also reads a machine's CPU use by systemd unit and by container, from each cgroup's cpu.stat, +so an overload outside every container can be named (novox/hq issue 741). It changes nothing and +declares no resources, so it can be assigned to every machine. ## Why it exists, and why here @@ -22,7 +24,7 @@ machine's resources a seat, these tools become its verbs. ## Tools -All four only read. +All five only read. | tool | what | replaces | |---|---|---| @@ -30,6 +32,7 @@ All four only read. | `disk_top` | Over a window. `by=unit` (the default) ranks systemd units and containers by bytes moved, from their cgroups. `by=container` ranks containers by name, and `by=process` ranks processes | `iotop` | | `disk_filesystems` | Each filesystem that stores data, said once however many places it is mounted: size, used and free GiB, used % as `df` counts it, inodes used % | `df -h`, `df -i` | | `disk_devices` | Each disk without sampling: model, size, whether it spins, scheduler, queue limit, what it is built on, and since boot the GiB read and written, the busy share and the wait per read and per write | `lsblk`, `cat /proc/diskstats` | +| `cpu_top` | Samples a window (`seconds`, default 3, at most 10). It answers the machine's CPU in use (% of all CPUs and as CPUs busy, from `/proc/stat`), then `by=unit` (the default) ranks systemd units and containers by CPU, from their cgroups' cpu.stat, busiest first (`limit`, default 15), with the busy time no unit accounts for; `by=container` ranks containers by name | `top`, `systemd-cgtop` | ### How each number is read diff --git a/modules/disk-load/cmd/disk-load/cgroups.go b/modules/disk-load/cmd/disk-load/cgroups.go index 79af261b..84f46e89 100644 --- a/modules/disk-load/cmd/disk-load/cgroups.go +++ b/modules/disk-load/cmd/disk-load/cgroups.go @@ -68,9 +68,10 @@ func (m *Machine) bottomDevices() map[string]bool { return out } -// unitIO is every unit's storage I/O, by its cgroup's path: the innermost .service or .scope, because -// io.stat counts a cgroup's descendants too and an outer unit would count them twice. -func (m *Machine) unitIO(counted map[string]bool) map[string]cgroupIO { +// unitPaths is every systemd unit's cgroup path under /sys/fs/cgroup: the innermost .service or +// .scope, because a cgroup's own stat file counts its descendants too and an outer unit would count +// them twice. Shared by every reader that walks the cgroup tree (unitIO here, unitCPU in cpu.go). +func (m *Machine) unitPaths() []string { base := m.path("/sys/fs/cgroup") var units []string filepath.WalkDir(base, func(p string, d fs.DirEntry, err error) error { @@ -89,11 +90,19 @@ func (m *Machine) unitIO(counted map[string]bool) map[string]cgroupIO { outer[dir] = true } } - out := map[string]cgroupIO{} + out := make([]string, 0, len(units)) for _, u := range units { - if outer[u] { - continue + if !outer[u] { + out = append(out, u) } + } + return out +} + +// unitIO is every unit's storage I/O, by its cgroup's path. +func (m *Machine) unitIO(counted map[string]bool) map[string]cgroupIO { + out := map[string]cgroupIO{} + for _, u := range m.unitPaths() { text := m.read("/sys/fs/cgroup/" + u + "/io.stat") if text == "" { continue diff --git a/modules/disk-load/cmd/disk-load/cpu.go b/modules/disk-load/cmd/disk-load/cpu.go new file mode 100644 index 00000000..efad5cb6 --- /dev/null +++ b/modules/disk-load/cmd/disk-load/cpu.go @@ -0,0 +1,176 @@ +package main + +import ( + "path/filepath" + "sort" + "strconv" + "strings" + "time" +) + +// cgroupCPU is a cgroup's processor time, from its cpu.stat. +type cgroupCPU struct { + usageUsec uint64 +} + +// parseCPUStat reads a cgroup's cpu.stat ("usage_usec 123\nuser_usec …\nsystem_usec …"), readable by +// every account, unlike /proc//stat for another account's process. +func parseCPUStat(s string) cgroupCPU { + var c cgroupCPU + for _, line := range strings.Split(s, "\n") { + k, v, ok := strings.Cut(line, " ") + if !ok { + continue + } + if k == "usage_usec" { + c.usageUsec, _ = strconv.ParseUint(strings.TrimSpace(v), 10, 64) + } + } + return c +} + +// unitCPU is every unit's processor time, by its cgroup's path. A cgroup whose cpu.stat cannot be +// read is left out, never an error for the whole answer. +func (m *Machine) unitCPU() map[string]cgroupCPU { + out := map[string]cgroupCPU{} + for _, u := range m.unitPaths() { + text := m.read("/sys/fs/cgroup/" + u + "/cpu.stat") + if text == "" { + continue + } + out[u] = parseCPUStat(text) + } + return out +} + +// MachineCPU is the machine's CPU in use over the window, from /proc/stat's cpu line. CPUsBusy is +// rounded only for display; raw, unexported, is the same count unrounded, for topCPU's share and +// unaccounted math (review round 1 of novox/hq issue 741: a small busy time that rounds CPUsBusy to +// 0.0 must not zero out every unit's share along with it). +type MachineCPU struct { + CPUs int `json:"cpus"` + BusyPct float64 `json:"busy_pct"` // % of all CPUs together + CPUsBusy float64 `json:"cpus_busy"` // as a count of CPUs, e.g. 7.5 of 12, rounded for display + raw float64 +} + +// machineCPUOver is the machine's CPU in use between two readings of /proc/stat: everything but idle +// and iowait, counted against the window's total ticks over every CPU. +func machineCPUOver(a, b cpuTimes) MachineCPU { + total := float64(b.total() - a.total()) + if total <= 0 { + return MachineCPU{CPUs: b.cpus} + } + idle := sub(b.idle, a.idle) + sub(b.iowait, a.iowait) + busy := total - idle + if busy < 0 { + busy = 0 + } + raw := busy / total * float64(b.cpus) + return MachineCPU{CPUs: b.cpus, BusyPct: round1(busy / total * 100), CPUsBusy: round1(raw), raw: raw} +} + +// CPUUsage is a unit's or a container's share of CPU between two readings of its cgroup's cpu.stat. +// Cgroup is its cgroup's path, relative to /sys/fs/cgroup, said beside Unit always (review round 1 of +// novox/hq issue 741): two cgroups can share a leaf name (an outer "init.scope" and a user session's +// "user.slice/user@1000.service/init.scope"), and the path is how they are told apart. +type CPUUsage struct { + Unit string `json:"unit,omitempty"` + Cgroup string `json:"cgroup,omitempty"` + Container string `json:"container,omitempty"` + CPUPct float64 `json:"cpu_pct"` // % of one CPU: 100 = one CPU busy the whole window + SharePct float64 `json:"share_pct"` // this unit's share of the machine's busy time + raw float64 +} + +// topCPU ranks the units, or the containers, with the most CPU time between two readings of +// cpu.stat, busiest first. unaccountedPct is the machine's busy time no unit accounts for, as a +// share of it (never below 0): the sum is always taken over every unit, whichever by ranks by, so a +// limit on the list never changes it. +func (m *Machine) topCPU(a, b map[string]cgroupCPU, window time.Duration, by string, limit int, cpusBusy float64) ([]CPUUsage, float64) { + windowUsec := window.Seconds() * 1e6 + type entry struct { + path, unit, container string + pct float64 + } + var entries []entry + var names map[string]string + var sumPct float64 + for path, end := range b { + start, ok := a[path] + if !ok { + continue + } + d := sub(end.usageUsec, start.usageUsec) + if d == 0 { + continue + } + pct := d / windowUsec * 100 + sumPct += pct + unit := filepath.Base(path) + var container string + if id := containerID.FindStringSubmatch("/" + path); id != nil { + if names == nil { + names = m.containerNames() + } + container = id[1][:12] + if n, ok := names[id[1]]; ok { + container = n + } + } + entries = append(entries, entry{path: path, unit: unit, container: container, pct: pct}) + } + + // Units are keyed by their cgroup's own path, never by its leaf name (filepath.Base): two cgroups can + // share a leaf name (an outer "init.scope" and a user session's nested one), and keying by the name + // alone would merge them into one row (review round 1 of novox/hq issue 741). Containers stay keyed + // by their name, as before. + groups := map[string]*CPUUsage{} + for _, e := range entries { + key := e.path + if by == "container" { + if e.container == "" { + continue + } + key = e.container + } + g, ok := groups[key] + if !ok { + g = &CPUUsage{Unit: e.unit, Cgroup: e.path, Container: e.container} + if by == "container" { + g.Unit, g.Cgroup = "", "" + } + groups[key] = g + } + g.raw += e.pct + } + out := make([]CPUUsage, 0, len(groups)) + for _, g := range groups { + out = append(out, *g) + } + sort.Slice(out, func(i, j int) bool { + if out[i].raw != out[j].raw { + return out[i].raw > out[j].raw + } + return out[i].Unit+out[i].Cgroup+out[i].Container < out[j].Unit+out[j].Cgroup+out[j].Container + }) + if len(out) > limit { + out = out[:limit] + } + for i := range out { + out[i].CPUPct = round1(out[i].raw) + if cpusBusy > 0 { + out[i].SharePct = round1(out[i].raw / (cpusBusy * 100) * 100) + } + } + + unaccounted := cpusBusy*100 - sumPct + if unaccounted < 0 { + unaccounted = 0 + } + unaccountedPct := 0.0 + if cpusBusy > 0 { + unaccountedPct = round1(unaccounted / (cpusBusy * 100) * 100) + } + return out, unaccountedPct +} diff --git a/modules/disk-load/cmd/disk-load/cpu_test.go b/modules/disk-load/cmd/disk-load/cpu_test.go new file mode 100644 index 00000000..5acc362a --- /dev/null +++ b/modules/disk-load/cmd/disk-load/cpu_test.go @@ -0,0 +1,182 @@ +package main + +import ( + "strings" + "testing" + "time" +) + +// aCPUHog is a machine whose CPU is 75 % busy (1.5 of 2 CPUs) over a 2 s window: a unit, another unit, +// and a container, each with a known usage_usec delta, leaving part of the machine's busy time +// unaccounted — the shape of issue 739, where 750 % of 12 CPUs went to processes outside containers. +func aCPUHog(t *testing.T) *fake { + f := newFake(t) + f.file("/proc/stat", "cpu 0 0 0 0 0 0 0 0 0 0\ncpu0 0 0 0 0 0 0 0 0 0 0\ncpu1 0 0 0 0 0 0 0 0 0 0\nintr 1\n") + f.then("/proc/stat", "cpu 100 0 50 50 0 0 0 0 0 0\ncpu0 50 0 25 25 0 0 0 0 0 0\ncpu1 50 0 25 25 0 0 0 0 0 0\nintr 2\n") + + zero := "usage_usec 0\nuser_usec 0\nsystem_usec 0\n" + cg := func(path string, usec int) { + f.file("/sys/fs/cgroup/"+path+"/cpu.stat", zero) + f.then("/sys/fs/cgroup/"+path+"/cpu.stat", "usage_usec "+itoa(usec)+"\nuser_usec "+itoa(usec)+"\nsystem_usec 0\n") + } + cg("system.slice/alpha.service", 1000000) // 50 % of one CPU over 2 s + cg("system.slice/beta.service", 400000) // 20 % + cg("system.slice/docker-"+store+".scope", 600000) // 30 %, named "cpu-hog" by the runtime + f.answers["docker ps --no-trunc --format {{.ID}}\t{{.Names}}"] = store + "\tcpu-hog\n" + return f +} + +func TestCPUTopRanksUnitsAndContainersByCPUOverTheWindow(t *testing.T) { + f := aCPUHog(t) + out, err := tool(t, f.machine(), "cpu_top").Run(map[string]any{"seconds": 2.0}) + if err != nil { + t.Fatal(err) + } + m := out.(map[string]any) + if m["by"] != "unit" || f.waited != 2*time.Second { + t.Errorf("the default: %+v, waited %s", m, f.waited) + } + cpu := m["cpu"].(MachineCPU) + if cpu.CPUs != 2 || cpu.BusyPct != 75 || cpu.CPUsBusy != 1.5 { + t.Errorf("machine cpu %+v", cpu) + } + top := m["top"].([]CPUUsage) + if len(top) != 3 { + t.Fatalf("three units moved: %+v", top) + } + if top[0].Unit != "alpha.service" || top[0].CPUPct != 50 || top[0].SharePct != 33.3 { + t.Errorf("busiest first: %+v", top[0]) + } + if top[1].Container != "cpu-hog" || top[1].CPUPct != 30 || top[1].SharePct != 20 { + t.Errorf("the container, named by the runtime: %+v", top[1]) + } + if top[2].Unit != "beta.service" || top[2].CPUPct != 20 || top[2].SharePct != 13.3 { + t.Errorf("the least busy: %+v", top[2]) + } + if m["unaccounted_pct"] != 33.3 { + t.Errorf("unaccounted: %+v (machine busy 150%% of one CPU, units sum to 100%%, 50%% of 150%% left over)", m["unaccounted_pct"]) + } +} + +func TestCPUTopByContainerNamesOnlyContainers(t *testing.T) { + out, err := tool(t, aCPUHog(t).machine(), "cpu_top").Run(map[string]any{"by": "container", "seconds": 2.0}) + if err != nil { + t.Fatal(err) + } + top := out.(map[string]any)["top"].([]CPUUsage) + if len(top) != 1 || top[0].Container != "cpu-hog" || top[0].Unit != "" || top[0].CPUPct != 30 { + t.Errorf("by container: %+v", top) + } + if _, err := tool(t, aCPUHog(t).machine(), "cpu_top").Run(map[string]any{"by": "process"}); err == nil { + t.Error("by=process was accepted; cpu_top only ranks units and containers") + } +} + +func TestCPUTopLimitsHowManyAreNamed(t *testing.T) { + out, err := tool(t, aCPUHog(t).machine(), "cpu_top").Run(map[string]any{"limit": 1.0, "seconds": 2.0}) + if err != nil { + t.Fatal(err) + } + m := out.(map[string]any) + top := m["top"].([]CPUUsage) + if len(top) != 1 || top[0].Unit != "alpha.service" { + t.Errorf("limit 1: %+v", top) + } + // The limit narrows the list, never the unaccounted share, which is taken over every unit. + if m["unaccounted_pct"] != 33.3 { + t.Errorf("unaccounted with limit 1: %+v", m["unaccounted_pct"]) + } +} + +func TestCPUTopLeavesOutACgroupThatAppearsOrVanishesWithinTheWindow(t *testing.T) { + f := aCPUHog(t) + f.then("/sys/fs/cgroup/system.slice/short-lived.service/cpu.stat", "usage_usec 500000\n") + out, err := tool(t, f.machine(), "cpu_top").Run(map[string]any{"seconds": 2.0}) + if err != nil { + t.Fatal(err) + } + top := out.(map[string]any)["top"].([]CPUUsage) + for _, u := range top { + if u.Unit == "short-lived.service" { + t.Errorf("a unit only in the second reading was counted: %+v", top) + } + } + if !strings.Contains(strings.Join(f.calls, ","), "docker ps") { + t.Errorf("the container runtime was never asked: %v", f.calls) + } +} + +// aTinyLoad is a machine with 12 CPUs, only 0.04 of them busy over a 2 s window (busy 1 of 300 total +// ticks): rounded to one decimal for display, CPUsBusy reads 0.0, but a unit using half of that busy +// time still has a non-zero share — review round 1's decision that share_pct and unaccounted_pct are +// computed from the unrounded busy time, never from the rounded CPUsBusy. +func aTinyLoad(t *testing.T) *fake { + f := newFake(t) + cpus := "cpu0 0 0 0 0 0 0 0 0 0 0\ncpu1 0 0 0 0 0 0 0 0 0 0\ncpu2 0 0 0 0 0 0 0 0 0 0\ncpu3 0 0 0 0 0 0 0 0 0 0\n" + + "cpu4 0 0 0 0 0 0 0 0 0 0\ncpu5 0 0 0 0 0 0 0 0 0 0\ncpu6 0 0 0 0 0 0 0 0 0 0\ncpu7 0 0 0 0 0 0 0 0 0 0\n" + + "cpu8 0 0 0 0 0 0 0 0 0 0\ncpu9 0 0 0 0 0 0 0 0 0 0\ncpu10 0 0 0 0 0 0 0 0 0 0\ncpu11 0 0 0 0 0 0 0 0 0 0\n" + f.file("/proc/stat", "cpu 0 0 0 0 0 0 0 0 0 0\n"+cpus) + f.then("/proc/stat", "cpu 1 0 0 299 0 0 0 0 0 0\n"+cpus) + + f.file("/sys/fs/cgroup/system.slice/tiny.service/cpu.stat", "usage_usec 0\n") + f.then("/sys/fs/cgroup/system.slice/tiny.service/cpu.stat", "usage_usec 60000\n") // 3 % of one CPU over 2 s + return f +} + +func TestCPUTopSharesAreNonZeroEvenWhenRoundedCPUsBusyIsZero(t *testing.T) { + out, err := tool(t, aTinyLoad(t).machine(), "cpu_top").Run(map[string]any{"seconds": 2.0}) + if err != nil { + t.Fatal(err) + } + m := out.(map[string]any) + cpu := m["cpu"].(MachineCPU) + if cpu.CPUs != 12 || cpu.CPUsBusy != 0 { + t.Fatalf("the machine 0.04 CPUs busy, rounded for display: %+v", cpu) + } + top := m["top"].([]CPUUsage) + if len(top) != 1 || top[0].Unit != "tiny.service" || top[0].CPUPct != 3 || top[0].SharePct != 75 { + t.Fatalf("a non-zero share from the unrounded busy time: %+v", top) + } + if m["unaccounted_pct"] != 25.0 { + t.Fatalf("unaccounted from the unrounded busy time: %+v (want 25, so shares plus unaccounted add to 100)", + m["unaccounted_pct"]) + } +} + +// TestCPUTopKeepsCgroupsWithTheSameUnitNameAsSeparateRows is review round 1's other decision: units are keyed +// by their cgroup path, never by filepath.Base, so two cgroups that both end in "init.scope" — one at the +// root, one under a user's session — are two rows, not one merged by their shared leaf name. +func TestCPUTopKeepsCgroupsWithTheSameUnitNameAsSeparateRows(t *testing.T) { + f := aCPUHog(t) + cg := func(path string, usec int) { + f.file("/sys/fs/cgroup/"+path+"/cpu.stat", "usage_usec 0\n") + f.then("/sys/fs/cgroup/"+path+"/cpu.stat", "usage_usec "+itoa(usec)+"\n") + } + cg("init.scope", 1000000) // the root init, 50 % of one CPU over 2 s + cg("user.slice/user@1000.service/init.scope", 400000) // a user session's init, 20 % + + out, err := tool(t, f.machine(), "cpu_top").Run(map[string]any{"seconds": 2.0, "limit": 10.0}) + if err != nil { + t.Fatal(err) + } + top := out.(map[string]any)["top"].([]CPUUsage) + var inits []CPUUsage + for _, u := range top { + if u.Unit == "init.scope" { + inits = append(inits, u) + } + } + if len(inits) != 2 { + t.Fatalf("two cgroups named init.scope, merged into %d row(s): %+v", len(inits), top) + } + byCgroup := map[string]CPUUsage{} + for _, u := range inits { + byCgroup[u.Cgroup] = u + } + if u, ok := byCgroup["init.scope"]; !ok || u.CPUPct != 50 { + t.Errorf("the root init.scope: %+v", byCgroup) + } + if u, ok := byCgroup["user.slice/user@1000.service/init.scope"]; !ok || u.CPUPct != 20 { + t.Errorf("the user session's init.scope: %+v", byCgroup) + } +} diff --git a/modules/disk-load/cmd/disk-load/tools.go b/modules/disk-load/cmd/disk-load/tools.go index b54bf21e..ed29c9c3 100644 --- a/modules/disk-load/cmd/disk-load/tools.go +++ b/modules/disk-load/cmd/disk-load/tools.go @@ -85,6 +85,57 @@ func Tools(m *Machine) []stdio.Tool { return out, nil }, }, + { + Name: "cpu_top", + Description: "What used the most CPU over a few seconds, from each unit's cgroup cpu.stat, which needs no root: the machine's " + + "CPU in use (as % of all CPUs and as CPUs busy, from /proc/stat), then by=unit (the default) each systemd unit and " + + "container ranked by CPU, busiest first, with the busy time no unit accounts for; by=container only the containers, " + + "by name. Replaces top, systemd-cgtop. (r)", + Input: map[string]any{ + "seconds": map[string]any{"type": "integer", "description": "how long to sample, in seconds (default 3, at most 10)"}, + "limit": map[string]any{"type": "integer", "description": "how many units or containers to name (default 15, at most 50)"}, + "by": map[string]any{"type": "string", "enum": []string{"unit", "container"}, "description": "what to rank (default unit)"}, + }, + Run: func(args map[string]any) (any, error) { + n, err := count(args, "seconds", 3, 1, 10) + if err != nil { + return nil, err + } + w := time.Duration(n) * time.Second + limit, err := count(args, "limit", 15, 1, 50) + if err != nil { + return nil, err + } + by := text(args, "by") + switch by { + case "": + by = "unit" + case "unit", "container": + default: + return nil, fmt.Errorf("by is unit or container, not %q", by) + } + ac, err := parseCPU(m.read("/proc/stat")) + if err != nil { + return nil, err + } + a := m.unitCPU() + m.Wait(w) + bc, err := parseCPU(m.read("/proc/stat")) + if err != nil { + return nil, err + } + b := m.unitCPU() + machine := machineCPUOver(ac, bc) + top, unaccounted := m.topCPU(a, b, w, by, limit, machine.raw) + return map[string]any{ + "window_s": w.Seconds(), + "by": by, + "cpu": machine, + "top": top, + "unaccounted_pct": unaccounted, + }, nil + }, + }, { Name: "disk_filesystems", Description: "The space of every filesystem that stores data (not proc, tmpfs or overlay layers): size, used and free in GiB, " + diff --git a/modules/disk-load/module.json b/modules/disk-load/module.json index 3b7ce911..e50253e6 100644 --- a/modules/disk-load/module.json +++ b/modules/disk-load/module.json @@ -5,7 +5,8 @@ "disk_load", "disk_top", "disk_filesystems", - "disk_devices" + "disk_devices", + "cpu_top" ], "build": { "artifacts": [ diff --git a/modules/mesh-delivery/cmd/mesh-delivery/holder.go b/modules/mesh-delivery/cmd/mesh-delivery/holder.go index a256007a..8cc6e8f7 100644 --- a/modules/mesh-delivery/cmd/mesh-delivery/holder.go +++ b/modules/mesh-delivery/cmd/mesh-delivery/holder.go @@ -312,19 +312,25 @@ func (h *Holder) PullUpdated(p PullEvent) { defer h.mu.Unlock() id := IDOf(p.repository(), p.HeadSHA) now := h.Now() + startedOver := "" if d := h.deliveries[id]; d != nil { - // Announced again: nothing new — but for a delivery whose verdict is out of date, the announcement is its - // check asked again against the trunk's head (the merge tool's ask, novox/hq ADR 0323 §5), and its bound - // runs from now (§7). The base head it was announced with is kept, for a check this owner asks after it. - if p.BaseSHA != "" && d.BaseSHA != p.BaseSHA { - d.BaseSHA = p.BaseSHA - h.keep(d) + if startedOver = startedOverFor(d, p.Number); startedOver == "" { + // Announced again: nothing new — but for a delivery whose verdict is out of date, the announcement is its + // check asked again against the trunk's head (the merge tool's ask, novox/hq ADR 0323 §5), and its bound + // runs from now (§7). The base head it was announced with is kept, for a check this owner asks after it. + if p.BaseSHA != "" && d.BaseSHA != p.BaseSHA { + d.BaseSHA = p.BaseSHA + h.keep(d) + } + if d.State == Proposed && d.outOfDate() != "" && d.AskedAgain == nil { + d.AskedAgain = &now + h.keep(d) + } + return } - if d.State == Proposed && d.outOfDate() != "" && d.AskedAgain == nil { - d.AskedAgain = &now - h.keep(d) - } - return + // A delivery stopped only because its pull request closed unmerged does not end its commit's delivery for + // another pull request with the same head (novox/hq issue 783): started over below, exactly as for a first + // announcement, under the same id — never a transition out of Stopped. } replaced := "" for _, old := range h.deliveries { @@ -338,11 +344,7 @@ func (h *Holder) PullUpdated(p PullEvent) { } } } - d := &Delivery{ID: id, Repository: p.repository(), Commit: p.HeadSHA, Number: p.Number, Title: p.Title, - Base: p.Base, Branch: p.Head, HTMLURL: p.HTMLURL, CloneURL: p.CloneURL, Paths: p.Paths, - PathsTruncated: p.PathsTruncated, Removed: p.Removed, ModuleDirs: p.ModuleDirs, ModuleDirsSaid: p.ModuleDirsSaid, - After: AfterIn(p.Body), Created: now, BaseSHA: p.BaseSHA} - t, err := Apply(d, EvAnnounced, false, Facts{Now: now, Why: fmt.Sprintf("%s#%d's head announced", d.Repository, d.Number)}) + d, t, err := h.freshAnnouncement(p, now, startedOver) if err != nil { h.refuse(err) return @@ -360,6 +362,34 @@ func (h *Holder) PullUpdated(p PullEvent) { } } +// startedOverFor is why a delivery already kept at a head's id is started over for a pull request's number rather +// than taken as announced again: only when it was stopped solely by an earlier, closed duplicate pull request with +// the same head — still Stopped, ClosedUnmerged, never merged, and under another number (novox/hq issue 783). +// Empty for the ordinary 'announced again' path, or for any other stop (a person's, a group's, its own close). +func startedOverFor(d *Delivery, number int) string { + if d != nil && d.State == Stopped && d.ClosedUnmerged && d.MergedAs == "" && d.Number != number { + return fmt.Sprintf("#%d with the same head closed unmerged", d.Number) + } + return "" +} + +// freshAnnouncement makes a delivery for p's head exactly as a first announcement would: its own id, proposed. +// startedOver, when not empty, is folded into the transition's Why, naming the pull request this replaces +// (novox/hq issue 783); the delivery it replaces is swapped out by the caller keeping the new one under the same +// id (keep), never by a transition out of its final state. +func (h *Holder) freshAnnouncement(p PullEvent, now time.Time, startedOver string) (*Delivery, Transition, error) { + d := &Delivery{ID: IDOf(p.repository(), p.HeadSHA), Repository: p.repository(), Commit: p.HeadSHA, Number: p.Number, + Title: p.Title, Base: p.Base, Branch: p.Head, HTMLURL: p.HTMLURL, CloneURL: p.CloneURL, Paths: p.Paths, + PathsTruncated: p.PathsTruncated, Removed: p.Removed, ModuleDirs: p.ModuleDirs, ModuleDirsSaid: p.ModuleDirsSaid, + After: AfterIn(p.Body), Created: now, BaseSHA: p.BaseSHA} + why := fmt.Sprintf("%s#%d's head announced", d.Repository, d.Number) + if startedOver != "" { + why = fmt.Sprintf("%s; %s", why, startedOver) + } + t, err := Apply(d, EvAnnounced, false, Facts{Now: now, Why: why}) + return d, t, err +} + // takePlaceOf puts a member's new head — a rebase, a push — in its group in the place of the head it superseded, // when the group is closed: a member of it already reached the trunk, and a closed group takes no new member. The // new head has its own check, and the group's composed check runs again for the new heads rather than holding the @@ -491,7 +521,29 @@ func (h *Holder) PullMerged(p PullEvent) { } now := h.Now() var ts []Transition - if d == nil { + if over := startedOverFor(d, p.Number); over != "" { + // A delivery stopped only by an earlier, closed duplicate pull request with the same head is not lost when + // its merge is heard without the open announcement either: it is started over for the merging pull request + // the same way an announcement would, before the merge is taken as usual (novox/hq issue 783). Heard this + // way — its own merge, with no open announcement of this pull request — it is made exactly as a first-heard + // merge in every field but its id: FromForge, and its check the forge's word on the head, so a long-past + // merge replayed from the bus's history still falls under the history rule (issue 313) and is not held for + // a person (review round 1). + fresh, t, err := h.freshAnnouncement(p, now, over) + if err != nil { + h.refuse(err) + return + } + fresh.FromForge = true + if s := p.HeadChecks["mesh/merge-gate"]; s != "" { + fresh.Check = &Verdict{At: now, Gate: verdictOfState(s), Summary: "the head's mesh/merge-gate, as the forge holds it", + Repo: verdictOfState(p.HeadChecks["mesh/repo-check"])} + } + d, ts = fresh, []Transition{t} + if d.Check != nil { + ts = append(ts, settle(d, Facts{Now: now})...) + } + } else if d == nil { head := p.HeadSHA if head == "" { head = p.MergeCommit diff --git a/modules/mesh-delivery/cmd/mesh-delivery/issue783_test.go b/modules/mesh-delivery/cmd/mesh-delivery/issue783_test.go new file mode 100644 index 00000000..c0dd6062 --- /dev/null +++ b/modules/mesh-delivery/cmd/mesh-delivery/issue783_test.go @@ -0,0 +1,151 @@ +package main + +import ( + "strings" + "testing" + "time" +) + +// novox/hq issue 783: mesh-catalog commit b5417c28 was pushed while #305 (from branch adr-359-mesh-mail-733) was +// open, so this owner recorded its delivery as #305's. #305 then closed unmerged (its base branch had been +// deleted), and #315, opened from the same branch and head, merged it as 0914a33f. The delivery stayed stopped +// ("the pull request closed unmerged"), though it recorded merged_as 0914a33f: a rule did not hold — a delivery +// keyed by commit follows the pull request that merges that commit, and a closed duplicate does not end it — and +// the walk that carried the merge waited 64 minutes for this owner's word until a person's `plans go` released it. +// +// A delivery stopped only because its pull request closed unmerged, with no merge recorded, does not end its +// commit's delivery for another pull request with the same head: it is started over, under the same id, for the +// pull request that announces or merges it next. + +const ( + head783, merge783 = "d305d305d305d305", "d315d315d315d315" + head783b, merge783b = "d306d306d306d306", "d316d316d316d316" + head783c = "d307d307d307d307" + head783d, merge783d = "d308d308d308d308", "d318d318d318d318" + catalog783 = "novox/mesh-catalog" + branch783 = "adr-359-mesh-mail-733" +) + +// TestADeliveryStoppedByAClosedDuplicateFollowsThePullRequestThatMergesItsCommit is the issue's own shape: #305 +// announced, closed unmerged (stopped); #315 announced from the same head, checked and readied; #315 merges it, +// and the walk that answers that merge is let go — the delivery follows #315, never stuck on #305's stop. +func TestADeliveryStoppedByAClosedDuplicateFollowsThePullRequestThatMergesItsCommit(t *testing.T) { + w := newWorld(t) + id := IDOf(catalog783, head783) + + w.h.PullUpdated(pr(catalog783, 305, head783, branch783, "src/a.go")) + w.h.PullClosed(pr(catalog783, 305, head783, branch783)) + if w.state(id) != Stopped { + t.Fatalf("#305 closed unmerged left it %s; want stopped", w.state(id)) + } + + w.h.PullUpdated(pr(catalog783, 315, head783, branch783, "src/a.go")) + if d := w.delivery(id); d.State == Stopped || d.Number != 315 { + t.Fatalf("#315, the same head, left it %s, Number %d; want started over for #315, not stopped", d.State, d.Number) + } + if d := w.delivery(id); !strings.Contains(d.Transitions[len(d.Transitions)-1].Why, "#305 with the same head closed unmerged") { + t.Fatalf("the start-over's transition names what it replaces: %q", d.Transitions[len(d.Transitions)-1].Why) + } + + w.h.Checked(verdict(catalog783, 315, head783, "pass")) + if d := w.delivery(id); d.State != Ready { + t.Fatalf("a passing verdict for #315 left it %s", d.State) + } + + walk := w.ctl.put(aWalk("plan-783", catalog783, merge783, true)) + w.h.WalkMoved(walk) + w.h.PullMerged(merged(catalog783, 315, head783, merge783)) + if d := w.delivery(id); d.State != Published || d.Number != 315 { + t.Fatalf("#315's merge left it %s, Number %d; want published under #315", d.State, d.Number) + } + + w.settleAll() + if len(w.ctl.delivers) != 1 || w.ctl.delivers[0] != "plan-783" { + t.Fatalf("the walk that answers #315's merge was not let go: %v", w.ctl.delivers) + } + if d := w.delivery(id); d.State != Delivering { + t.Fatalf("let go, it is %s; want delivering", d.State) + } +} + +// TestAMergeWithoutAnAnnouncementStartsOverAStoppedDuplicateToo is decision 4: the merging pull request is never +// announced on its own (heard only as `pull.merged`) — the start-over happens in PullMerged itself, not only in +// PullUpdated, so a merge heard without the open announcement is not lost either. +func TestAMergeWithoutAnAnnouncementStartsOverAStoppedDuplicateToo(t *testing.T) { + w := newWorld(t) + id := IDOf(catalog783, head783b) + + w.h.PullUpdated(pr(catalog783, 306, head783b, branch783+"-b", "src/b.go")) + w.h.PullClosed(pr(catalog783, 306, head783b, branch783+"-b")) + if w.state(id) != Stopped { + t.Fatalf("#306 closed unmerged left it %s; want stopped", w.state(id)) + } + + walk := w.ctl.put(aWalk("plan-783b", catalog783, merge783b, true)) + w.h.WalkMoved(walk) + m := merged(catalog783, 316, head783b, merge783b) + m.HeadChecks = map[string]string{"mesh/merge-gate": "success", "mesh/repo-check": "success"} + w.h.PullMerged(m) + if d := w.delivery(id); d.State == Stopped || d.Number != 316 { + t.Fatalf("#316's merge, heard with no announcement of its own, left it %s, Number %d; want started over for #316", + d.State, d.Number) + } + + w.settleAll() + if len(w.ctl.delivers) != 1 || w.ctl.delivers[0] != "plan-783b" { + t.Fatalf("the walk that answers #316's merge was not let go: %v", w.ctl.delivers) + } +} + +// TestAStoppedDeliveryAnnouncedAgainByItsOwnPullRequestStaysStopped guards decision 3's "any other stopped +// delivery ... is untouched": the same pull request announced again after its own close is the ordinary +// 'announced again' path, never a start-over, and stays stopped. +func TestAStoppedDeliveryAnnouncedAgainByItsOwnPullRequestStaysStopped(t *testing.T) { + w := newWorld(t) + id := IDOf(catalog783, head783c) + + w.h.PullUpdated(pr(catalog783, 307, head783c, branch783+"-c", "src/c.go")) + w.h.PullClosed(pr(catalog783, 307, head783c, branch783+"-c")) + if w.state(id) != Stopped { + t.Fatalf("#307 closed unmerged left it %s; want stopped", w.state(id)) + } + + w.h.PullUpdated(pr(catalog783, 307, head783c, branch783+"-c", "src/c.go")) + if d := w.delivery(id); d.State != Stopped || d.Number != 307 { + t.Fatalf("#307 announced again by itself left it %s, Number %d; want it to stay stopped", d.State, d.Number) + } +} + +// TestAMergeHeardLongAfterStartingOverAStoppedDuplicateIsHistoryLikeAnyOtherReplayedMerge is review round 1's +// decision: a delivery started over from a merge heard without its open announcement (as in decision 4, above) +// takes FromForge exactly as a first-heard merge does, so one replayed from the bus's history long after it +// happened still falls under the history rule (novox/hq issue 313) and is never held for a person. +func TestAMergeHeardLongAfterStartingOverAStoppedDuplicateIsHistoryLikeAnyOtherReplayedMerge(t *testing.T) { + w := newWorld(t) + id := IDOf(catalog783, head783d) + + w.h.PullUpdated(pr(catalog783, 308, head783d, branch783+"-d", "src/d.go")) + w.h.PullClosed(pr(catalog783, 308, head783d, branch783+"-d")) + if w.state(id) != Stopped { + t.Fatalf("#308 closed unmerged left it %s; want stopped", w.state(id)) + } + + m := merged(catalog783, 318, head783d, merge783d) + m.MergedAt = w.now.Add(-historyLate - time.Minute).UTC().Format(time.RFC3339) + w.h.PullMerged(m) + + d := w.delivery(id) + if d.State == Stopped || d.Number != 318 { + t.Fatalf("#318's merge, heard long after it happened with no announcement of its own, left it %s, Number %d; "+ + "want started over for #318", d.State, d.Number) + } + if !d.FromForge { + t.Fatal("a merge heard without its own announcement is made from the forge's word: FromForge is false") + } + if d.History == "" { + t.Fatalf("a merge heard %s after it happened is history, never held for a person; History is empty", historyLate+time.Minute) + } + if d.State == Held { + t.Fatalf("history is never held for a person; state is %s, HeldWhy %q", d.State, d.HeldWhy) + } +} diff --git a/modules/mesh-mail/README.md b/modules/mesh-mail/README.md index be92c413..c5104514 100644 --- a/modules/mesh-mail/README.md +++ b/modules/mesh-mail/README.md @@ -2,7 +2,9 @@ One module holds the operator's mail accounts, watches them, and says each change of a message on the bus for the modules the operator grants it to (novox/hq ADR 0359, research 049 document 02). One instance, on the control-node. -This is the first stage: accounts of the kind `imap`, read-only. Nothing here flags, moves, deletes, drafts or sends. +Reading is the first stage: accounts of the kind `imap`, read-only — nothing here flags, moves, deletes or drafts. +Sending is `mail_send` alone (novox/hq ADR 0361 §2): it sends nothing itself, and reaches an account's outgoing +server only once the operator approves it on a channel that proves who answered. ## What it does @@ -32,12 +34,13 @@ This is the first stage: accounts of the kind `imap`, read-only. Nothing here fl "folders": ["optional: the folders watched besides the inbox; every folder when left out"]}} ``` -`outgoing` is read and never used in this stage. A second `imap` account on another server is one more entry and one -more password; no code changes. `user` signs in under another name than the address. +`outgoing` is the server `mail_send` sends through: host, port, and `security` `tls` (465, implicit) or `starttls` +(587). A second `imap` account on another server is one more entry and one more password; no code changes. `user` +signs in under another name than the address. -`grants` — per account, who may read its mail: a module's name (on every machine), `.`, or -`person.`. Nobody by default. It restricts the stream `MAIL` on the bus (a person's entry grants nothing there) -and the tools `mail_search` and `mail_read`. +`grants` — per account, who may read its mail, and who may send from it: a module's name (on every machine), +`.`, or `person.`. Nobody by default. It restricts the stream `MAIL` on the bus (a person's +entry grants nothing there) and the tools `mail_search`, `mail_read` and `mail_send` — the same check for all three. The password of an account is the own secret `mail-credential-`, given at the desk; the mesh never makes one. @@ -49,8 +52,12 @@ The password of an account is the own secret `mail-credential-`, given | `mail_search` | messages of the accounts the caller is granted, by every word of a query, newest first, no body | | `mail_read` | one message's metadata and text, or one attachment's bytes, to a caller granted its account | | `mail_health` | healthy; unhealthy naming the account and why (connection down, or no answer for five minutes); waits naming `mail-credential-` while its password is not given | +| `mail_send` | raises an ask on the operator channel (the account, the recipients, the subject, the first lines) for a caller granted the account; answers at once with the ask's id. Sends nothing itself | +| `mail_send_status` | where a send stands, by its ask id: waiting, sent, declined, ended without a choice, or not proven | -The caller is the runtime's word, never an argument. +The caller is the runtime's word, never an argument. `mail_send` sends through the account's outgoing server only +once the operator's Approve comes back on a channel that proves who answered, and once: a warrant delivered twice +never sends twice (novox/hq ADR 0361 §2). ## Its processes diff --git a/modules/mesh-mail/cmd/mesh-mail/config_test.go b/modules/mesh-mail/cmd/mesh-mail/config_test.go index 1fb38430..21acbb3a 100644 --- a/modules/mesh-mail/cmd/mesh-mail/config_test.go +++ b/modules/mesh-mail/cmd/mesh-mail/config_test.go @@ -28,8 +28,8 @@ func TestTheFirstStagesAccountIsReadAndItsSecretNamed(t *testing.T) { } // A second account of the kind imap on another server — Gmail with an app password, another member of the family — -// is configuration only: no code changes for it. Each account may also say its outgoing server, which this stage -// reads and never uses (it sends nothing). +// is configuration only: no code changes for it. Each account may also say its outgoing server, which mail_send +// uses to send through it (novox/hq ADR 0361 §1). func TestASecondIMAPAccountIsConfigurationOnly(t *testing.T) { got, err := ParseAccounts([]byte(`{"accounts": { "mesh": {"kind": "imap", "address": "jschoubben@novox.be", "incoming": {"host": "mail.novox.be", "port": 993, "security": "tls"}}, diff --git a/modules/mesh-mail/cmd/mesh-mail/imap_test.go b/modules/mesh-mail/cmd/mesh-mail/imap_test.go index 6a62105e..47d9d53d 100644 --- a/modules/mesh-mail/cmd/mesh-mail/imap_test.go +++ b/modules/mesh-mail/cmd/mesh-mail/imap_test.go @@ -209,8 +209,11 @@ func TestTheKeeperWaitsForThePasswordAndSaysARefusal(t *testing.T) { }) } -// It only reads (ADR 0359 §5): no code of the module, but its tests, issues a command that flags, moves, copies, -// deletes, appends, creates, renames or sends, and every folder is opened read-only and every body peeked. +// It only reads a mailbox (ADR 0359 §5): no code of the module, but its tests, issues a command that flags, moves, +// copies, deletes, appends, creates, renames or otherwise writes to one over IMAP, and every folder is opened +// read-only and every body peeked. mail_send is the one exception the operator's own Approve carves out (ADR 0361 +// §2): it never touches a mailbox either — it speaks SMTP, in smtp.go alone, to the account's outgoing server, and +// only once the operator approved it (send.go). func TestNothingButTheTestsWritesToAMailbox(t *testing.T) { files, err := filepath.Glob("*.go") if err != nil { @@ -218,7 +221,9 @@ func TestNothingButTheTestsWritesToAMailbox(t *testing.T) { } writes := regexp.MustCompile(`([A-Za-z_][A-Za-z0-9_]*)\.(Store|Move|Copy|Expunge|UIDExpunge|Append|Create|Delete|Rename|Subscribe|Unsubscribe|SetMetadata|SetQuota)\(`) for _, f := range files { - if strings.HasSuffix(f, "_test.go") { + if strings.HasSuffix(f, "_test.go") || f == "send.go" || f == "smtp.go" { + // Neither touches a mailbox: send.go keeps the `sends` state on the bus, never IMAP; smtp.go speaks SMTP + // to the account's outgoing server, never to a mailbox, and only on the operator's Approve (ADR 0361 §2). continue } raw, err := os.ReadFile(f) @@ -226,12 +231,15 @@ func TestNothingButTheTestsWritesToAMailbox(t *testing.T) { t.Fatal(err) } for _, m := range writes.FindAllSubmatch(raw, -1) { - if string(m[1]) != "os" { // a file of the module's own, never a mailbox + // os is a file of the module's own, never a mailbox; stdio is the bus SDK (main.go's hear subscribes to + // the router's own decided subject, exactly as mesh-knowledge does, novox/hq issue 790 review round 1) + // — its Subscribe has nothing to do with an IMAP mailbox's. + if string(m[1]) != "os" && string(m[1]) != "stdio" { t.Errorf("%s calls %s, which writes to a mailbox", f, m[0]) } } if strings.Contains(string(raw), "net/smtp") { - t.Errorf("%s speaks SMTP: this stage sends nothing", f) + t.Errorf("%s speaks SMTP: only smtp.go does, and only on the operator's Approve (ADR 0361 §2)", f) } } src, _ := os.ReadFile("imap.go") diff --git a/modules/mesh-mail/cmd/mesh-mail/main.go b/modules/mesh-mail/cmd/mesh-mail/main.go index 34ae3094..02d4fffa 100644 --- a/modules/mesh-mail/cmd/mesh-mail/main.go +++ b/modules/mesh-mail/cmd/mesh-mail/main.go @@ -1,8 +1,10 @@ // mesh-mail: the operator's mail accounts, watched, and each mail said on the bus (novox/hq ADR 0359). One instance, -// on the control-node. The first stage holds the mesh's own server, read-only: no flag, move, delete, draft or send. +// on the control-node. Reading is read-only: no flag, move, delete or draft. mail_send is the one way out (novox/hq +// ADR 0361 §2): it sends nothing itself, and reaches an account's outgoing server only on the operator's Approve. // -// mesh-mail the bundle the node's runtime launches: the tools mail_accounts, mail_search, mail_read and -// mail_health, and the events the keeper wrote, said on the bus as the module +// mesh-mail the bundle the node's runtime launches: the tools mail_accounts, mail_search, mail_read, +// mail_health, mail_send and mail_send_status, and the events the keeper wrote, said on the +// bus as the module // mesh-mail prepare the preparation step: the index's schema brought to this version's // mesh-mail keep acts: watches every account (IDLE on the inbox, a look at the other folders every five // minutes), keeps the index and the attachment store, and writes each change's event beside it @@ -17,6 +19,7 @@ package main import ( "context" + "encoding/json" "fmt" "os" "os/signal" @@ -25,6 +28,7 @@ import ( "time" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/asks" ) // Module is this module's name. @@ -176,14 +180,51 @@ func serve(env Env) { return NewS3(facts), nil }, } + b.Logf = logf p := &Publisher{Index: b.Index, Emit: func(key string, body any) error { return stdio.Emit(key, body) }, Logf: logf} go p.Run(context.Background()) + go hear(b) + go sweepSends(b) if err := stdio.Serve("", b.Tools()); err != nil { logf("%v", err) os.Exit(1) } } +// hear subscribes to the router's word on this module's own asks, asked again until the runtime takes it, exactly +// as mesh-knowledge does. Each warrant heard goes to Heard, the same Decide that mail_send_status uses: an +// approved send goes out as soon as the operator answers it, never only when something later calls +// mail_send_status (novox/hq ADR 0361 §2, review round 1 of issue 790). +func hear(b *Bundle) { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + time.Sleep(wait) + err := stdio.Subscribe(asks.Seat+".decided."+Module, func(e stdio.Envelope) error { + var w asks.Warrant + if json.Unmarshal(e.Body, &w) == nil { + b.Heard(w) + } + return nil + }) + if err == nil { + logf("[mesh-mail] hears the operator's answers") + return + } + logf("[mesh-mail] does not hear the operator's answers yet (%v); asking again in %s", err, wait) + } +} + +// sweepSends reads every send still undecided once a minute and asks the router's own record for its answer, so +// an Approve answered while this module could not hear it directly is still acted on (novox/hq issue 790, review +// round 1). +func sweepSends(b *Bundle) { + tick := time.NewTicker(SweepEvery) + defer tick.Stop() + for { + b.Sweep(time.Now()) + <-tick.C + } +} + // Publisher says the events the keeper wrote, oldest first, and removes each once the bus took it. One the bus did not // take stays, and is said again on the next round: delivery is at least once, and a consumer reads the uid and the // folder to know a repeat. diff --git a/modules/mesh-mail/cmd/mesh-mail/manifest_test.go b/modules/mesh-mail/cmd/mesh-mail/manifest_test.go index 128300cb..d6563479 100644 --- a/modules/mesh-mail/cmd/mesh-mail/manifest_test.go +++ b/modules/mesh-mail/cmd/mesh-mail/manifest_test.go @@ -78,7 +78,7 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { for _, tool := range (&Bundle{}).Tools() { names = append(names, tool.Name) } - if !reflect.DeepEqual(m.Tools, names) || !reflect.DeepEqual(names, []string{"mail_accounts", "mail_search", "mail_read", "mail_health"}) { + if !reflect.DeepEqual(m.Tools, names) || !reflect.DeepEqual(names, []string{"mail_accounts", "mail_search", "mail_read", "mail_health", "mail_send", "mail_send_status"}) { t.Fatalf("tools %v, the code serves %v", m.Tools, names) } // Its events go to its own stream MAIL, kept a week, each account's part heard only by the consumers the trusted diff --git a/modules/mesh-mail/cmd/mesh-mail/send.go b/modules/mesh-mail/cmd/mesh-mail/send.go new file mode 100644 index 00000000..f05f828a --- /dev/null +++ b/modules/mesh-mail/cmd/mesh-mail/send.go @@ -0,0 +1,643 @@ +package main + +// mail_send (novox/hq ADR 0361 §2): the tool that replaces the no-send rule of ADR 0359 §5, for sending alone. It +// sends nothing itself. A caller granted the account (the same check as mail_read, ADR 0359 §4) asks for one send; +// this raises an ask on the operator channel (novox/hq ADR 0259) showing the account, the recipients, the subject +// and the first lines of the body, and the tool returns at once with the ask's id. The mail reaches the account's +// outgoing server only on the operator's Approve, on a channel that proves who answered, and once: the id is kept +// with the decision it was given, so a warrant delivered twice is acted on once (asks.Warrant.For and +// asks.Option.Performs hold the decision to exactly the request that was asked). A Decline, or no answer before it +// expires, sends nothing, which mail_send_status says. The password is read only to sign in, and the body only to +// send; neither is ever logged. +// +// The pattern — publish an ask under this module's own name, compose it again from what was kept to hold the +// router's warrant, act once on a compare-and-set claim — follows mesh-knowledge's own approval of a record +// (cmd/mesh-knowledge/operator.go) and claude-code's proposals and questions (cmd/claude-code/proposals.go, +// questions.go). + +import ( + "crypto/rand" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "net/mail" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/asks" +) + +// SendsState is the module's state the sends are kept in. +const SendsState = "sends" + +// The two answers, both at the level approve, so only a channel that proves who answered can carry either. +const ( + OptionSend = "send" + OptionDecline = "decline" +) + +// SendAskLasts is how long the operator has to answer: a day less a margin, under the router's bound for an approval. +const SendAskLasts = asks.ApproveLasts - 10*time.Minute + +// StaleSendAfter is how long a send may read "sending" before mail_send_status stops calling it "waiting": past +// this, the claim before the SMTP call is old enough that the send may or may not have gone out, and is never +// retried automatically (novox/hq issue 790, review round 1). +const StaleSendAfter = 10 * time.Minute + +// SweepEvery is how often the sweep reads every send still undecided and asks the router's own record for its +// answer, so an Approve answered while this module did not hear it directly (down, or not yet subscribed) is +// still acted on (novox/hq issue 790, review round 1). +const SweepEvery = time.Minute + +// undecidedIndexKey is a reserved key in the `sends` state (never a send id, which always starts "send-") that +// keeps the ids of the sends still undecided, so the sweep can list only those instead of every send ever kept +// (novox/hq issue 811). It stays in the same declared state rather than a state of its own, which a manifest +// change would need. +const undecidedIndexKey = "undecided-index" + +// undecidedIndex is the undecidedIndexKey document: the ids of the sends still undecided. +type undecidedIndex struct { + IDs []string `json:"ids"` +} + +// addUndecided adds id to the undecided index, retrying its compare-and-set write on a race. A stored index that +// cannot be parsed is never overwritten (which would silently drop every id it already named): it is reported as +// an error naming the key instead (novox/hq issue 811, review round 3). +func addUndecided(store SendStore, id string) error { + for { + raw, rev, ok, err := store.Get(undecidedIndexKey) + if err != nil { + return err + } + var idx undecidedIndex + if ok { + if err := json.Unmarshal(raw, &idx); err != nil { + return fmt.Errorf("the undecided index %q cannot be parsed: %w", undecidedIndexKey, err) + } + for _, existing := range idx.IDs { + if existing == id { + return nil + } + } + } + idx.IDs = append(idx.IDs, id) + if !ok { + if _, err := store.Create(undecidedIndexKey, idx); err != nil { + if errors.Is(err, stdio.ErrChanged) { + continue + } + return err + } + return nil + } + if _, err := store.Update(undecidedIndexKey, idx, rev); err != nil { + if errors.Is(err, stdio.ErrChanged) { + continue + } + return err + } + return nil + } +} + +// removeUndecided drops id from the undecided index, retrying its compare-and-set write on a race. A send never +// read by the index (already absent) is left as it is. A stored index that cannot be parsed is never overwritten: +// it is reported as an error naming the key instead (novox/hq issue 811, review round 3). +func removeUndecided(store SendStore, id string) error { + for { + raw, rev, ok, err := store.Get(undecidedIndexKey) + if err != nil || !ok { + return err + } + var idx undecidedIndex + if err := json.Unmarshal(raw, &idx); err != nil { + return fmt.Errorf("the undecided index %q cannot be parsed: %w", undecidedIndexKey, err) + } + at := -1 + for i, existing := range idx.IDs { + if existing == id { + at = i + break + } + } + if at == -1 { + return nil + } + idx.IDs = append(idx.IDs[:at], idx.IDs[at+1:]...) + if _, err := store.Update(undecidedIndexKey, idx, rev); err != nil { + if errors.Is(err, stdio.ErrChanged) { + continue + } + return err + } + return nil + } +} + +// undecidedIDs is every id the undecided index currently names, for the sweep to read. An index that cannot be +// parsed is reported as an error naming the key, not silently read as empty (novox/hq issue 811, review round 3). +func undecidedIDs(store SendStore) ([]string, error) { + raw, _, ok, err := store.Get(undecidedIndexKey) + if err != nil || !ok { + return nil, err + } + var idx undecidedIndex + if err := json.Unmarshal(raw, &idx); err != nil { + return nil, fmt.Errorf("the undecided index %q cannot be parsed: %w", undecidedIndexKey, err) + } + return idx.IDs, nil +} + +// The bounds of what mail_send takes. +const ( + MostRecipients = 20 + MostSubject = 200 + MostSendBody = 1 << 20 // 1 MiB; only the first ShownBodyChars reach the operator's ask + ShownBodyChars = 300 +) + +// SendRequest is one send asked of the operator, as the `sends` state keeps it. Every field the ask composes from +// is here, so the decision is held to exactly what was asked. +type SendRequest struct { + ID string `json:"id"` + Account string `json:"account"` + To []string `json:"to"` + Subject string `json:"subject"` + Body string `json:"body"` + Caller string `json:"caller"` + At time.Time `json:"at"` + // Decided is set once, and never changed again: the ask is acted on once (ADR 0361 §2). + Decided *SendDecision `json:"decided,omitempty"` +} + +// SendDecision is what became of one send, kept once and read back by mail_send_status. +type SendDecision struct { + At time.Time `json:"at"` + Outcome string `json:"outcome"` + Warrant string `json:"warrant,omitempty"` + Error string `json:"error,omitempty"` +} + +// bodyDigest is the digest an act binds in place of the body itself: never the body, which the act's digest is +// published and read back on the bus. +func bodyDigest(body string) string { + sum := sha256.Sum256([]byte(body)) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +// Act is what an answer binds: this request, whole, and the option chosen. +func (s SendRequest) Act(option string) asks.Act { + return asks.Act{"verb": Module + ".send", "option": option, "send": s.ID, "account": s.Account, + "to": strings.Join(s.To, ","), "subject": s.Subject, "body-digest": bodyDigest(s.Body), + "caller": s.Caller, "at": s.At.UTC().Format(time.RFC3339Nano)} +} + +// firstChars is at most n runes of s, marked when it was cut. +func firstChars(s string, n int) string { + r := []rune(s) + if len(r) <= n { + return s + } + return string(r[:n]) + " […cut]" +} + +// Ask is the ask for this request, composed from it alone: the account, the recipients, the subject and the first +// lines of the body (novox/hq ADR 0361 §2). +func (s SendRequest) Ask() asks.Ask { + to := strings.Join(s.To, ", ") + headline := fmt.Sprintf("Send mail from %s?", s.Account) + if len([]rune(headline)) > asks.HeadlineLength { + headline = "Send mail?" + } + explanation := fmt.Sprintf("From %s to %s: %q.", s.Account, to, s.Subject) + if len([]rune(explanation)) > asks.SummaryLength { + explanation = fmt.Sprintf("From %s: %q.", s.Account, s.Subject) + } + details := fmt.Sprintf("Asked by %s through the mesh-mail module, at %s UTC.\nAccount: %s\nTo: %s\nSubject: %s\n\n%s", + s.Caller, s.At.UTC().Format("15:04 on 2 Jan"), s.Account, to, s.Subject, firstChars(s.Body, ShownBodyChars)) + a := asks.Ask{ID: s.ID, Headline: headline, Explanation: explanation, Details: details, + Who: asks.Operator, Expires: s.At.Add(SendAskLasts), OnExpiry: "nothing is sent"} + for _, o := range []struct{ id, label, does string }{ + {OptionSend, "Send", "the mail is sent through " + s.Account + "'s outgoing server"}, + {OptionDecline, "Decline", "nothing is sent"}, + } { + binds, _ := asks.ActDigest(s.Act(o.id)) + a.Options = append(a.Options, asks.Option{ID: o.id, Label: o.label, Does: o.does, Level: asks.Approve, Binds: binds}) + } + return a +} + +func newSendID() string { + var b [8]byte + _, _ = rand.Read(b[:]) + return "send-" + hex.EncodeToString(b[:]) +} + +// ---- the bus, as a send reaches it ---------------------------------------------------------------------------- + +// AskBus is the operator channel as this module reaches it: publish an ask under its own name, and read the +// router's own record of one of its asks. +type AskBus interface { + Publish(subject string, body any, id string) error + Record(id string) (*stdio.StateEntry, error) +} + +type sdkAskBus struct{} + +func (sdkAskBus) Publish(subject string, body any, id string) error { + return stdio.SeatPublish(subject, body, id) +} +func (sdkAskBus) Record(id string) (*stdio.StateEntry, error) { return stdio.SeatRecord("", id) } + +// SendStore is the `sends` state: a value with its revision, created only when absent, updated only at the revision +// read — so of two writers only the first writes (the compare-and-set a decision is claimed by). +type SendStore interface { + Get(key string) (json.RawMessage, uint64, bool, error) + Create(key string, value any) (uint64, error) + Update(key string, value any, revision uint64) (uint64, error) + Keys() ([]string, error) +} + +// keptSends is the `sends` state on the bus. +type keptSends struct{ s stdio.KeptState } + +func (k keptSends) Get(key string) (json.RawMessage, uint64, bool, error) { + e, err := k.s.Get(key) + if err != nil || e == nil { + return nil, 0, false, err + } + return e.Value, e.Revision, true, nil +} +func (k keptSends) Create(key string, value any) (uint64, error) { return k.s.Create(key, value) } +func (k keptSends) Update(key string, value any, revision uint64) (uint64, error) { + return k.s.Update(key, value, revision) +} +func (k keptSends) Keys() ([]string, error) { return k.s.Keys() } + +func getSend(store SendStore, id string) (SendRequest, uint64, bool, error) { + raw, rev, ok, err := store.Get(id) + if err != nil || !ok { + return SendRequest{}, 0, false, err + } + var req SendRequest + if json.Unmarshal(raw, &req) != nil || req.ID != id { + return SendRequest{}, 0, false, nil + } + return req, rev, true, nil +} + +// routerRecord is the router's record of one ask, as far as this module reads it. +type routerRecord struct { + State string `json:"state"` + Warrant *asks.Warrant `json:"warrant"` +} + +// ---- asking ---------------------------------------------------------------------------------------------------- + +// SendInput is one send a caller asks for: the account is already the caller's to use (the grants check, the same +// as mail_read's, is the tool's to make before this is called). +type SendInput struct { + Account Account + To []string + Subject string + Body string + Caller string +} + +// cleanAddresses trims and drops empty entries. +func cleanAddresses(in []string) []string { + var out []string + for _, a := range in { + if a = strings.TrimSpace(a); a != "" { + out = append(out, a) + } + } + return out +} + +// hasControlChar is true when s holds a byte a mail header must never carry: CR, LF, any other ASCII control +// character below 0x20, or DEL (0x7f) — exactly what a header injection needs (novox/hq issue 790, review round 1). +func hasControlChar(s string) bool { + for i := 0; i < len(s); i++ { + if s[i] < 0x20 || s[i] == 0x7f { + return true + } + } + return false +} + +// AskMailSend raises an ask for one send and keeps it, under the module's own name. It never sends: the mail +// reaches the server only once the operator's Approve is read back and held to this request (Decide). It answers +// at once, with the ask's id; mail_send_status says where it stands. +func AskMailSend(in SendInput, store SendStore, bus AskBus, now time.Time) (map[string]any, error) { + refuse := func(why string) (map[string]any, error) { return nil, errors.New(why + ". Nothing was sent") } + to := cleanAddresses(in.To) + switch { + case len(to) == 0: + return refuse("to names no recipient") + case len(to) > MostRecipients: + return refuse(fmt.Sprintf("to names %d recipients, more than %d", len(to), MostRecipients)) + } + for _, addr := range to { + if hasControlChar(addr) { + return refuse(fmt.Sprintf("%q is not a mail address: it holds a control character", addr)) + } + if _, err := mail.ParseAddress(addr); err != nil { + return refuse(fmt.Sprintf("%q is not a mail address", addr)) + } + } + subject := strings.TrimSpace(in.Subject) + switch { + case subject == "": + return refuse("subject is empty") + case hasControlChar(subject): + return refuse("the subject holds a control character") + case len([]rune(subject)) > MostSubject: + return refuse(fmt.Sprintf("the subject is %d characters, more than %d", len([]rune(subject)), MostSubject)) + } + switch { + case strings.TrimSpace(in.Body) == "": + return refuse("body is empty") + case len(in.Body) > MostSendBody: + return refuse(fmt.Sprintf("the body is %d bytes, more than %d", len(in.Body), MostSendBody)) + } + if in.Account.Outgoing == nil { + return refuse(fmt.Sprintf("the account %s has no outgoing server configured", in.Account.Name)) + } + req := SendRequest{ID: newSendID(), Account: in.Account.Name, To: to, Subject: subject, Body: in.Body, + Caller: in.Caller, At: now.UTC()} + ask := req.Ask() + if err := ask.Check(now); err != nil { + return refuse(err.Error()) + } + if _, err := store.Create(req.ID, req); err != nil { + return nil, fmt.Errorf("the send could not be kept, so the operator was not asked: %w. Nothing was sent", err) + } + if err := addUndecided(store, req.ID); err != nil { + // Nothing was published yet, so refusing here costs nothing: an ask the sweep could never see is never + // sent (novox/hq issue 811, review round 3). + return nil, fmt.Errorf("the send could not be made visible to the sweep, so the operator was not asked: %w. Nothing was sent", err) + } + if err := bus.Publish(asks.AskSubject(Module), ask, ask.ID); err != nil { + return nil, fmt.Errorf("the operator could not be asked (%v). Nothing was sent; mail_send_status %s says so", err, req.ID) + } + return map[string]any{ + "ask": req.ID, + "status": "waiting", + "asked": "\"" + ask.Headline + "\", to " + to[0] + onwards(to), + "from": in.Account.Name, + "expires": ask.Expires.Format(time.RFC3339), + "next": "mail_send_status with this ask id says where it stands; nothing is sent until the operator's Approve, " + + "and nothing on Decline or at expiry", + }, nil +} + +func onwards(to []string) string { + if len(to) <= 1 { + return "" + } + return fmt.Sprintf(" and %d more", len(to)-1) +} + +// ---- deciding --------------------------------------------------------------------------------------------------- + +// AccountLookup resolves an account by name, as the settings currently hold it (not as it stood when it was asked: +// the account's server may have changed since, and its password is read fresh), or found false. +type AccountLookup func(name string) (Account, bool) + +// SendActor performs the one send a chosen option authorises, resolving the account's own secret itself. It is +// called at most once per request (Decide claims the request first). +type SendActor func(a Account, to []string, subject, body string, now time.Time) error + +// settleSend writes a decision onto a kept send (claiming it, or ending it outright) and, once the write lands, +// drops it from the undecided index: an index entry is never left behind once the record it points to carries a +// decision (novox/hq issue 811). On a race lost to another writer, it reads back whatever that writer decided. +func settleSend(store SendStore, req SendRequest, rev uint64, d SendDecision) (SendDecision, uint64, error) { + next := req + next.Decided = &d + newRev, err := store.Update(req.ID, next, rev) + if err != nil { + if errors.Is(err, stdio.ErrChanged) { + if again, _, ok, err2 := getSend(store, req.ID); err2 == nil && ok && again.Decided != nil { + return *again.Decided, 0, nil + } + } + return d, 0, err + } + _ = removeUndecided(store, req.ID) // best-effort (see removeUndecided) + return d, newRev, nil +} + +// Decide holds one closed ask's warrant to the request it answers (asks.Warrant.For, asks.Option.Performs), and — +// on Approve, exactly once — performs the send. A request already decided is never decided again: decide answers +// its kept decision at once, and two warrants for the same request (a redelivery, or two readers racing) have only +// the first claim the request by a compare-and-set write; the second finds it already claimed or decided and sends +// nothing. +func Decide(store SendStore, lookup AccountLookup, send SendActor, w asks.Warrant, now time.Time) (SendDecision, error) { + req, rev, ok, err := getSend(store, w.Ask) + if err != nil { + return SendDecision{}, err + } + if !ok { + return SendDecision{}, fmt.Errorf("no send %q is kept", w.Ask) + } + if req.Decided != nil { + return *req.Decided, nil + } + if w.Asker != Module { + return SendDecision{}, fmt.Errorf("the warrant is for %s's ask %s, not %s's", w.Asker, w.Ask, Module) + } + settle := func(d SendDecision) (SendDecision, error) { + final, _, err := settleSend(store, req, rev, d) + return final, err + } + if w.Outcome != asks.OutcomeChosen { + outcome := "ended: " + string(w.Outcome) + if w.Words != "" { + outcome += " (" + w.Words + ")" + } + return settle(SendDecision{At: now, Outcome: outcome}) + } + ask := req.Ask() + option, err := w.For(Module, ask) + if err == nil { + err = option.Performs(req.Act(option.ID)) + } + if err != nil { + return settle(SendDecision{At: now, Outcome: "refused: " + err.Error()}) + } + switch option.ID { + case OptionDecline: + return settle(SendDecision{At: now, Outcome: "declined", Warrant: w.Says()}) + case OptionSend: + return decideSend(store, lookup, send, req, rev, w, now) + default: + return settle(SendDecision{At: now, Outcome: "refused: the option " + option.ID + " is not one mail_send offers"}) + } +} + +// decideSend claims the request by a compare-and-set write before it sends, so a redelivered warrant finds it +// already claimed (or decided) and acts on nothing: a person's Approve is performed once, never retried on a +// transient failure (the failure is kept in the decision, not hidden). +func decideSend(store SendStore, lookup AccountLookup, send SendActor, req SendRequest, rev uint64, w asks.Warrant, now time.Time) (SendDecision, error) { + claimed := req + claim := SendDecision{At: now, Outcome: "sending", Warrant: w.Says()} + claimedDecision, newRev, err := settleSend(store, req, rev, claim) + if err != nil { + return SendDecision{}, err + } + if newRev == 0 { + // Lost the claim race: another writer already decided this send. + return claimedDecision, nil + } + claimed.Decided = &claim + acc, found := lookup(req.Account) + var final SendDecision + switch { + case !found: + final = SendDecision{At: now, Outcome: "refused: the account no longer exists", Warrant: w.Says()} + default: + if sendErr := send(acc, req.To, req.Subject, req.Body, now); sendErr != nil { + final = SendDecision{At: now, Outcome: "failed: " + sendErr.Error(), Warrant: w.Says(), Error: sendErr.Error()} + } else { + final = SendDecision{At: now, Outcome: "sent", Warrant: w.Says()} + } + } + claimed.Decided = &final + if _, err := store.Update(req.ID, claimed, newRev); err != nil { + // The claim stands either way: a person's Approve is acted on once. The caller logs this; it is not retried. + return final, fmt.Errorf("the send %s was %s, and that could not be kept (%w)", req.ID, final.Outcome, err) + } + return final, nil +} + +// ---- reading the status ----------------------------------------------------------------------------------------- + +// ReadSendStatus is where a send asked with mail_send stands, by its ask id: waiting for the operator, sent, +// declined, ended without a choice (expired, refused, cancelled), or not proven. Read from the router's own +// record; when the router's word is a close one that was not decided yet, this is also where a send is first +// decided (and so sent), exactly once (Decide). +func ReadSendStatus(store SendStore, bus AskBus, lookup AccountLookup, send SendActor, id string, now time.Time) (map[string]any, error) { + trimmed := strings.TrimSpace(id) + req, rev, ok, err := getSend(store, trimmed) + if err != nil { + return nil, fmt.Errorf("the send %q cannot be read now (%v); ask again", trimmed, err) + } + if !ok { + return nil, fmt.Errorf("no send %q is kept: the id is the one mail_send gave", trimmed) + } + out := map[string]any{"ask": req.ID, "account": req.Account, "to": req.To, "subject": req.Subject, + "asked": req.At.Format(time.RFC3339)} + if req.Decided != nil { + reportDecision(out, *req.Decided, now) + return out, nil + } + entry, err := bus.Record(req.ID) + if err != nil { + return nil, fmt.Errorf("the router's record of the ask cannot be read now (%v); ask again", err) + } + if entry == nil { + // The router never recorded this ask at all (its Publish failed, or it was never heard): once its asked + // time plus SendAskLasts has passed, it never will be, so it is settled as ended rather than left + // "waiting" forever (novox/hq issue 811). + if now.Sub(req.At) > SendAskLasts { + d, _, err := settleSend(store, req, rev, SendDecision{At: now, Outcome: "ended: never heard"}) + if err != nil { + return nil, fmt.Errorf("the send %q could not be settled as ended (%v); ask again", trimmed, err) + } + reportDecision(out, d, now) + return out, nil + } + out["status"], out["words"] = "waiting", "not yet recorded by the router; nothing is sent until the operator answers" + return out, nil + } + var rec routerRecord + if json.Unmarshal(entry.Value, &rec) != nil { + out["status"], out["words"] = "not proven", "the router's record of it cannot be read" + return out, nil + } + if rec.State == "" || rec.State == "open" || rec.Warrant == nil { + out["status"], out["words"] = "waiting", "the operator has not answered yet" + return out, nil + } + d, err := Decide(store, lookup, send, *rec.Warrant, now) + if err != nil { + out["status"], out["words"] = "not proven", err.Error() + return out, nil + } + reportDecision(out, d, now) + return out, nil +} + +// reportDecision turns a kept decision into what mail_send_status answers: a status word a caller can branch on, +// and the words that explain it. now is when it is asked, not when the send was decided: a claim still "sending" +// longer than StaleSendAfter with no final write is "not proven" (the send may or may not have gone out), never +// "waiting" forever — and it is never retried automatically either way (novox/hq issue 790, review round 1). +func reportDecision(out map[string]any, d SendDecision, now time.Time) { + out["at"] = d.At.Format(time.RFC3339) + switch { + case d.Outcome == "sent": + out["status"], out["words"] = "sent", "sent on the operator's Approve ("+d.Warrant+")" + case d.Outcome == "declined": + out["status"], out["words"] = "declined", "nothing was sent: the operator declined ("+d.Warrant+")" + case d.Outcome == "sending" && now.Sub(d.At) > StaleSendAfter: + out["status"] = "not proven" + out["words"] = "not proven: the send may or may not have gone out; the operator's Approve was being acted " + + "on at " + d.At.Format(time.RFC3339) + ", with no final word since" + case d.Outcome == "sending": + out["status"], out["words"] = "waiting", "the operator's Approve is being acted on; ask again shortly" + case d.Outcome == "ended: never heard": + out["status"], out["words"] = "ended", "its ask to the operator was never heard, so it was not sent" + case strings.HasPrefix(d.Outcome, "failed: "): + out["status"], out["words"] = "failed", "nothing reached the server: "+strings.TrimPrefix(d.Outcome, "failed: ") + case strings.HasPrefix(d.Outcome, "ended: "): + out["status"], out["words"] = "ended", "nothing was sent: "+strings.TrimPrefix(d.Outcome, "ended: ") + case strings.HasPrefix(d.Outcome, "refused: "): + out["status"], out["words"] = "not proven", "nothing was sent: "+strings.TrimPrefix(d.Outcome, "refused: ") + default: + out["status"], out["words"] = "ended", d.Outcome + } +} + +// ---- heard directly, and swept ------------------------------------------------------------------------------- + +// Heard acts on one warrant the router's own subject delivered to this module directly (main.go's hear), through +// the same Decide that mail_send_status uses: an approved send goes out as soon as the operator answers, never +// only when something later happens to call mail_send_status (novox/hq ADR 0361 §2, review round 1 of issue 790). +func (b *Bundle) Heard(w asks.Warrant) { + if _, err := Decide(b.sends(), b.lookupAccount, b.sendActor(), w, b.now()); err != nil { + if b.Logf != nil { + b.Logf("[mesh-mail] the warrant for the send %s could not be acted on: %v", w.Ask, err) + } + } +} + +// Sweep reads only the sends the undecided index still names, and asks the router's own record for each one's +// answer, the same path mail_send_status takes (ReadSendStatus): an Approve recorded while this module did not +// hear it directly (down, or not yet subscribed) is still acted on, once. Unlike reading every send ever kept, its +// work grows only with the sends still undecided, never with every one a module has ever sent (novox/hq ADR 0361 +// §2, review round 1 of issue 790; the index, issue 811). +func (b *Bundle) Sweep(now time.Time) { + store := b.sends() + ids, err := undecidedIDs(store) + if err != nil { + if b.Logf != nil { + b.Logf("[mesh-mail] the undecided index cannot be read: %v", err) + } + return + } + for _, id := range ids { + req, _, ok, err := getSend(store, id) + if err != nil || !ok || req.Decided != nil { + continue + } + if _, err := ReadSendStatus(store, b.askBus(), b.lookupAccount, b.sendActor(), id, now); err != nil { + if b.Logf != nil { + b.Logf("[mesh-mail] sweeping the send %s: %v", id, err) + } + } + } +} diff --git a/modules/mesh-mail/cmd/mesh-mail/send_test.go b/modules/mesh-mail/cmd/mesh-mail/send_test.go new file mode 100644 index 00000000..056400c9 --- /dev/null +++ b/modules/mesh-mail/cmd/mesh-mail/send_test.go @@ -0,0 +1,745 @@ +package main + +// mail_send (novox/hq ADR 0361 §2): an ungranted caller is refused before anything is asked; a granted caller's ask +// holds the account, the recipients, the subject and the first lines; the operator's Approve sends through an +// in-process fake SMTP server exactly once, even delivered twice; a Decline sends nothing. + +import ( + "bufio" + "encoding/json" + "errors" + "fmt" + "net" + "net/smtp" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/asks" +) + +// ---- a fake `sends` state and a fake operator channel, as a test keeps them -------------------------------------- + +type memSends struct { + mu sync.Mutex + vals map[string]json.RawMessage + revs map[string]uint64 + n uint64 + + // failKey/failErr: when failKey matches, Create and Update report failErr instead of writing, for a test to + // simulate a store whose index write fails (novox/hq issue 811, review round 3). + failKey string + failErr error +} + +func newMemSends() *memSends { + return &memSends{vals: map[string]json.RawMessage{}, revs: map[string]uint64{}} +} + +func (m *memSends) Get(key string) (json.RawMessage, uint64, bool, error) { + m.mu.Lock() + defer m.mu.Unlock() + v, ok := m.vals[key] + if !ok { + return nil, 0, false, nil + } + return v, m.revs[key], true, nil +} +func (m *memSends) Create(key string, value any) (uint64, error) { + m.mu.Lock() + defer m.mu.Unlock() + if m.failErr != nil && key == m.failKey { + return 0, m.failErr + } + if _, ok := m.vals[key]; ok { + return 0, stdio.ErrChanged + } + return m.put(key, value) +} +func (m *memSends) Update(key string, value any, revision uint64) (uint64, error) { + m.mu.Lock() + defer m.mu.Unlock() + if m.failErr != nil && key == m.failKey { + return 0, m.failErr + } + if revision == 0 || m.revs[key] != revision { + return 0, stdio.ErrChanged + } + return m.put(key, value) +} +func (m *memSends) put(key string, value any) (uint64, error) { + raw, err := json.Marshal(value) + if err != nil { + return 0, err + } + m.n++ + m.vals[key], m.revs[key] = raw, m.n + return m.n, nil +} +func (m *memSends) Keys() ([]string, error) { + m.mu.Lock() + defer m.mu.Unlock() + out := make([]string, 0, len(m.vals)) + for k := range m.vals { + out = append(out, k) + } + return out, nil +} + +// sendRouter is the operator channel's holder as this module reaches it, kept by a test: what was asked, and the +// router's own record of each ask. +type sendRouter struct { + mu sync.Mutex + published map[string]asks.Ask + records map[string]json.RawMessage + publishErr error + recordCalls map[string]int +} + +func newSendRouter() *sendRouter { + return &sendRouter{published: map[string]asks.Ask{}, records: map[string]json.RawMessage{}, recordCalls: map[string]int{}} +} + +func (r *sendRouter) Publish(subject string, body any, id string) error { + r.mu.Lock() + defer r.mu.Unlock() + if r.publishErr != nil { + return r.publishErr + } + raw, _ := json.Marshal(body) + var a asks.Ask + _ = json.Unmarshal(raw, &a) + r.published[a.ID] = a + r.records[a.ID], _ = json.Marshal(map[string]any{"state": "open"}) + return nil +} + +// recordCallsFor is how many times Record was asked about id since the last resetRecordCalls, for a test to check +// the sweep read only the sends still undecided (novox/hq issue 811). +func (r *sendRouter) recordCallsFor(id string) int { + r.mu.Lock() + defer r.mu.Unlock() + return r.recordCalls[id] +} + +func (r *sendRouter) resetRecordCalls() { + r.mu.Lock() + defer r.mu.Unlock() + r.recordCalls = map[string]int{} +} + +func (r *sendRouter) Record(id string) (*stdio.StateEntry, error) { + r.mu.Lock() + defer r.mu.Unlock() + r.recordCalls[id]++ + raw, ok := r.records[id] + if !ok { + return nil, nil + } + return &stdio.StateEntry{Key: id, Value: raw}, nil +} + +// close ends an ask with a warrant, as the router records it. +func (r *sendRouter) close(w asks.Warrant) { + r.mu.Lock() + defer r.mu.Unlock() + r.records[w.Ask], _ = json.Marshal(map[string]any{"state": string(w.Outcome), "warrant": w}) +} + +// chose is the warrant for a person's choice of an option of the ask as the router took it. +func (r *sendRouter) chose(id, option string, at time.Time) asks.Warrant { + r.mu.Lock() + a := r.published[id] + r.mu.Unlock() + o, _ := a.Option(option) + return asks.Warrant{Ask: id, Asker: Module, Outcome: asks.OutcomeChosen, Option: o.ID, Label: o.Label, Level: o.Level, + By: &asks.Person{Who: asks.Operator, Kind: "telegram", Identity: "42", Verified: "user id verified"}, + Channel: "telegram", Proofs: []string{"P1"}, At: at, AskDigest: a.Digest()} +} + +func (r *sendRouter) published1() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.published) +} + +// ---- an in-process fake SMTP server, speaking the protocol over a real socket ------------------------------------- + +type receivedMail struct { + from string + to []string + data string +} + +type fakeSMTPServer struct { + ln net.Listener + + mu sync.Mutex + received []receivedMail + refuse bool // when true, DATA is refused: a transient server error +} + +func newFakeSMTPServer(t *testing.T) *fakeSMTPServer { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + f := &fakeSMTPServer{ln: ln} + go f.serve() + t.Cleanup(func() { _ = ln.Close() }) + return f +} + +func (f *fakeSMTPServer) addr() string { return f.ln.Addr().String() } + +// dialer reaches the fake server directly, plain: only a test skips the TLS the mesh's own dialer always uses. +func (f *fakeSMTPServer) dialer() SMTPDialer { + addr := f.addr() + return func(Server) (*smtp.Client, error) { return smtp.Dial(addr) } +} + +func (f *fakeSMTPServer) count() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.received) +} + +func (f *fakeSMTPServer) last() receivedMail { + f.mu.Lock() + defer f.mu.Unlock() + if len(f.received) == 0 { + return receivedMail{} + } + return f.received[len(f.received)-1] +} + +func (f *fakeSMTPServer) serve() { + for { + conn, err := f.ln.Accept() + if err != nil { + return + } + go f.handle(conn) + } +} + +func (f *fakeSMTPServer) handle(conn net.Conn) { + defer conn.Close() + r := bufio.NewReader(conn) + reply := func(code int, msg string) { fmt.Fprintf(conn, "%d %s\r\n", code, msg) } + reply(220, "fake.local ESMTP") + var from string + var to []string + for { + line, err := r.ReadString('\n') + if err != nil { + return + } + line = strings.TrimRight(line, "\r\n") + upper := strings.ToUpper(line) + switch { + case strings.HasPrefix(upper, "EHLO"), strings.HasPrefix(upper, "HELO"): + reply(250, "fake.local") + case strings.HasPrefix(upper, "AUTH"): + reply(235, "authenticated") + case strings.HasPrefix(upper, "MAIL FROM:"): + from = addrIn(line) + reply(250, "OK") + case strings.HasPrefix(upper, "RCPT TO:"): + to = append(to, addrIn(line)) + reply(250, "OK") + case upper == "DATA": + f.mu.Lock() + refuse := f.refuse + f.mu.Unlock() + if refuse { + reply(451, "temporary failure") + continue + } + reply(354, "End data with .") + var b strings.Builder + for { + dl, err := r.ReadString('\n') + if err != nil { + return + } + if dl == ".\r\n" || dl == ".\n" { + break + } + if strings.HasPrefix(dl, "..") { + dl = dl[1:] + } + b.WriteString(dl) + } + f.mu.Lock() + f.received = append(f.received, receivedMail{from: from, to: append([]string{}, to...), data: b.String()}) + f.mu.Unlock() + reply(250, "OK: queued") + from, to = "", nil + case upper == "RSET": + reply(250, "OK") + from, to = "", nil + case upper == "QUIT": + reply(221, "bye") + return + default: + reply(500, "unrecognized command") + } + } +} + +func addrIn(line string) string { + start, end := strings.Index(line, "<"), strings.Index(line, ">") + if start >= 0 && end > start { + return line[start+1 : end] + } + return line +} + +// ---- a bundle over an account with an outgoing server --------------------------------------------------------- + +// sendTestBundle is a bundle over one account, "mesh", granted to "triage" alone, with an outgoing server of the +// caller's choosing (a fake SMTP server's, in most tests) and its password already given. +func sendTestBundle(t *testing.T, outgoing Server) (*Bundle, *sendRouter, *memSends) { + t.Helper() + dir := t.TempDir() + env := Env{Accounts: filepath.Join(dir, "accounts.json"), Grants: filepath.Join(dir, "grants.json"), Secrets: dir} + writeFile(t, env.Accounts, fmt.Sprintf(`{"accounts": {"mesh": {"kind": "imap", "address": "jschoubben@novox.be", + "incoming": {"host": "mail.novox.be", "port": 993, "security": "tls"}, + "outgoing": {"host": %q, "port": %d, "security": %q}}}}`, outgoing.Host, outgoing.Port, outgoing.Security)) + writeFile(t, env.Grants, `{"grants": {"mesh": ["triage"]}}`) + giveSecret(t, dir, Account{Name: "mesh"}, "the-password") + router, sends := newSendRouter(), newMemSends() + b := &Bundle{Env: env, Now: func() time.Time { return judgedAt }, Asks: router, Sends: sends} + return b, router, sends +} + +var sendArgs = map[string]any{"account": "mesh", "to": []any{"ops@example.org", "lead@example.org"}, + "subject": "The nightly gate is red", "body": "The merge gate failed on the resolver change.\nSee the job log."} + +// (1) An ungranted caller is refused, and no ask is raised. +func TestMailSendRefusesAnUngrantedCallerAndRaisesNoAsk(t *testing.T) { + b, router, _ := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"}) + rt := startBundle(t, b.Tools()) + text, isErr := rt.call(SendTool, "novox.other", sendArgs) + if !isErr || !strings.Contains(text, "refused") || !strings.Contains(text, "mesh") { + t.Fatalf("%v %s", isErr, text) + } + if n := router.published1(); n != 0 { + t.Fatalf("an ask was raised for an ungranted caller: %d", n) + } +} + +// (2) A granted caller raises one ask whose text holds the account, the recipients, the subject and the first lines. +func TestMailSendAsksWithTheAccountRecipientsSubjectAndFirstLines(t *testing.T) { + b, router, _ := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"}) + rt := startBundle(t, b.Tools()) + text, isErr := rt.call(SendTool, "novox.triage", sendArgs) + if isErr { + t.Fatalf("%s", text) + } + var out map[string]any + if json.Unmarshal([]byte(text), &out) != nil { + t.Fatalf("not JSON: %s", text) + } + id, _ := out["ask"].(string) + if id == "" || out["status"] != "waiting" { + t.Fatalf("%s", text) + } + if n := router.published1(); n != 1 { + t.Fatalf("published %d asks, not 1", n) + } + a := router.published[id] + if err := a.Check(judgedAt); err != nil { + t.Errorf("the ask does not pass the asks contract: %v", err) + } + whole := a.Headline + "\n" + a.Explanation + "\n" + a.Details + for _, want := range []string{"mesh", "ops@example.org", "lead@example.org", "The nightly gate is red", "The merge gate failed on the resolver change"} { + if !strings.Contains(whole, want) { + t.Errorf("the ask does not say %q:\n%s", want, whole) + } + } + for _, o := range a.Options { + if o.Level != asks.Approve || o.Binds == "" { + t.Errorf("the option %s is %s binding %q: both answers are an approval bound to this send", o.ID, o.Level, o.Binds) + } + } +} + +// (3) On Approve, the fake SMTP server receives exactly one message, even if the warrant is delivered twice. +func TestOnApproveTheFakeSMTPServerReceivesExactlyOneMessageEvenDeliveredTwice(t *testing.T) { + fake := newFakeSMTPServer(t) + b, router, sends := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"}) + b.Dial = fake.dialer() + rt := startBundle(t, b.Tools()) + + text, isErr := rt.call(SendTool, "novox.triage", sendArgs) + if isErr { + t.Fatalf("%s", text) + } + var out map[string]any + _ = json.Unmarshal([]byte(text), &out) + id := out["ask"].(string) + + w := router.chose(id, OptionSend, judgedAt.Add(time.Minute)) + router.close(w) + + lookup := b.lookupAccount + actor := b.sendActor() + if _, err := Decide(sends, lookup, actor, w, judgedAt.Add(2*time.Minute)); err != nil { + t.Fatalf("deciding: %v", err) + } + if n := fake.count(); n != 1 { + t.Fatalf("the fake server received %d message(s) after one Approve", n) + } + // The warrant delivered a second time: still exactly one message. + if _, err := Decide(sends, lookup, actor, w, judgedAt.Add(3*time.Minute)); err != nil { + t.Fatalf("deciding again: %v", err) + } + if n := fake.count(); n != 1 { + t.Fatalf("the fake server received %d message(s) after the warrant was delivered twice", n) + } + got := fake.last() + if got.from != "jschoubben@novox.be" || len(got.to) != 2 || got.to[0] != "ops@example.org" || got.to[1] != "lead@example.org" { + t.Fatalf("received %+v", got) + } + if !strings.Contains(got.data, "The nightly gate is red") || !strings.Contains(got.data, "The merge gate failed on the resolver change") { + t.Fatalf("the message sent: %q", got.data) + } + + // mail_send_status reports it sent, and reports the same thing again without sending a third time. + text, isErr = rt.call(SendStatusTool, "novox.triage", map[string]any{"id": id}) + if isErr || !strings.Contains(text, `"status":"sent"`) { + t.Fatalf("%v %s", isErr, text) + } + if n := fake.count(); n != 1 { + t.Fatalf("mail_send_status sent again: %d", n) + } +} + +// (4) On Decline, nothing is sent, and the status says declined. +func TestOnDeclineNothingIsSentAndTheStatusSaysDeclined(t *testing.T) { + fake := newFakeSMTPServer(t) + b, router, sends := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"}) + b.Dial = fake.dialer() + rt := startBundle(t, b.Tools()) + + text, isErr := rt.call(SendTool, "novox.triage", sendArgs) + if isErr { + t.Fatalf("%s", text) + } + var out map[string]any + _ = json.Unmarshal([]byte(text), &out) + id := out["ask"].(string) + + w := router.chose(id, OptionDecline, judgedAt.Add(time.Minute)) + router.close(w) + if _, err := Decide(sends, b.lookupAccount, b.sendActor(), w, judgedAt.Add(2*time.Minute)); err != nil { + t.Fatalf("deciding: %v", err) + } + if n := fake.count(); n != 0 { + t.Fatalf("the fake server received %d message(s) on Decline", n) + } + + text, isErr = rt.call(SendStatusTool, "novox.triage", map[string]any{"id": id}) + if isErr || !strings.Contains(text, `"status":"declined"`) { + t.Fatalf("%v %s", isErr, text) + } +} + +// An account with no outgoing server is refused before anything is asked. +func TestMailSendRefusesAnAccountWithNoOutgoingServer(t *testing.T) { + dir := t.TempDir() + env := Env{Accounts: filepath.Join(dir, "accounts.json"), Grants: filepath.Join(dir, "grants.json"), Secrets: dir} + writeFile(t, env.Accounts, stageOne) // no outgoing server + writeFile(t, env.Grants, `{"grants": {"mesh": ["triage"]}}`) + router := newSendRouter() + b := &Bundle{Env: env, Now: func() time.Time { return judgedAt }, Asks: router, Sends: newMemSends()} + rt := startBundle(t, b.Tools()) + text, isErr := rt.call(SendTool, "novox.triage", sendArgs) + if !isErr || !strings.Contains(text, "no outgoing server") { + t.Fatalf("%v %s", isErr, text) + } + if n := router.published1(); n != 0 { + t.Fatalf("an ask was raised for an account with no outgoing server: %d", n) + } +} + +// (5) An approved send goes out once the warrant reaches the module through the hear handler alone — never +// through mail_send_status (novox/hq ADR 0361 §2, review round 1 of issue 790). +func TestAnApprovedSendGoesOutOnceWithNoStatusCall(t *testing.T) { + fake := newFakeSMTPServer(t) + b, router, _ := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"}) + b.Dial = fake.dialer() + rt := startBundle(t, b.Tools()) + + text, isErr := rt.call(SendTool, "novox.triage", sendArgs) + if isErr { + t.Fatalf("%s", text) + } + var out map[string]any + _ = json.Unmarshal([]byte(text), &out) + id := out["ask"].(string) + + w := router.chose(id, OptionSend, judgedAt.Add(time.Minute)) + router.close(w) + + b.Heard(w) // the warrant delivered through the hear handler only + if n := fake.count(); n != 1 { + t.Fatalf("the fake server received %d message(s) after the warrant was heard once", n) + } + b.Heard(w) // the same warrant delivered a second time + if n := fake.count(); n != 1 { + t.Fatalf("the fake server received %d message(s) after the warrant was heard twice", n) + } +} + +// (6) The sweep acts on a send the operator approved while this module was never told directly: it reads the +// router's own record, the same path mail_send_status takes (novox/hq issue 790, review round 1). +func TestSweepActsOnAnApprovedSendItWasNotToldAboutDirectly(t *testing.T) { + fake := newFakeSMTPServer(t) + b, router, _ := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"}) + b.Dial = fake.dialer() + rt := startBundle(t, b.Tools()) + + text, isErr := rt.call(SendTool, "novox.triage", sendArgs) + if isErr { + t.Fatalf("%s", text) + } + var out map[string]any + _ = json.Unmarshal([]byte(text), &out) + id := out["ask"].(string) + + w := router.chose(id, OptionSend, judgedAt.Add(time.Minute)) + router.close(w) // only the router's record holds the answer; mesh-mail is never told directly + + b.Sweep(judgedAt.Add(2 * time.Minute)) + if n := fake.count(); n != 1 { + t.Fatalf("the sweep did not act on the approved send: %d message(s)", n) + } + b.Sweep(judgedAt.Add(3 * time.Minute)) + if n := fake.count(); n != 1 { + t.Fatalf("a second sweep sent again: %d message(s)", n) + } +} + +// (7) A send claimed "sending" longer than StaleSendAfter with no final write reads as "not proven", never +// "waiting" forever (novox/hq issue 790, review round 1). +func TestAStaleSendingClaimIsNotProvenNotWaiting(t *testing.T) { + b, _, sends := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"}) + rt := startBundle(t, b.Tools()) + + text, isErr := rt.call(SendTool, "novox.triage", sendArgs) + if isErr { + t.Fatalf("%s", text) + } + var out map[string]any + _ = json.Unmarshal([]byte(text), &out) + id := out["ask"].(string) + + req, rev, ok, err := getSend(sends, id) + if err != nil || !ok { + t.Fatalf("%v %v", err, ok) + } + claim := SendDecision{At: judgedAt.Add(-11 * time.Minute), Outcome: "sending", Warrant: "W1"} + req.Decided = &claim + if _, err := sends.Update(req.ID, req, rev); err != nil { + t.Fatalf("%v", err) + } + + text, isErr = rt.call(SendStatusTool, "novox.triage", map[string]any{"id": id}) + if isErr { + t.Fatalf("%s", text) + } + if strings.Contains(text, `"status":"waiting"`) { + t.Fatalf("still reports waiting past the stale bound: %s", text) + } + if !strings.Contains(text, `"status":"not proven"`) { + t.Fatalf("does not report not proven once the claim is stale: %s", text) + } +} + +// (8) The subject and every address are refused when they hold a control character: no header injection +// (novox/hq issue 790, review round 1). +func TestMailSendRefusesAControlCharacterInTheSubjectOrAnAddress(t *testing.T) { + b, _, _ := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"}) + rt := startBundle(t, b.Tools()) + + bySubject := map[string]any{"account": "mesh", "to": []any{"ops@example.org"}, + "subject": "Report\r\nBcc: attacker@example.org", "body": "body"} + text, isErr := rt.call(SendTool, "novox.triage", bySubject) + if !isErr || !strings.Contains(text, "control character") { + t.Fatalf("a subject with a control character was not refused: %v %s", isErr, text) + } + + byAddress := map[string]any{"account": "mesh", "to": []any{"ops@example.org\r\nBcc: attacker@example.org"}, + "subject": "Report", "body": "body"} + text, isErr = rt.call(SendTool, "novox.triage", byAddress) + if !isErr || !strings.Contains(text, "control character") { + t.Fatalf("an address with a control character was not refused: %v %s", isErr, text) + } +} + +// (9) Rcpt takes the parsed address alone; the To header takes the formatted address (novox/hq issue 790, review +// round 1). +func TestSendUsesTheParsedAddressForRcptAndTheFormattedAddressForTheToHeader(t *testing.T) { + fake := newFakeSMTPServer(t) + b, router, _ := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"}) + b.Dial = fake.dialer() + rt := startBundle(t, b.Tools()) + + args := map[string]any{"account": "mesh", "to": []any{"Ops Team "}, + "subject": "Status", "body": "body"} + text, isErr := rt.call(SendTool, "novox.triage", args) + if isErr { + t.Fatalf("%s", text) + } + var out map[string]any + _ = json.Unmarshal([]byte(text), &out) + id := out["ask"].(string) + + w := router.chose(id, OptionSend, judgedAt.Add(time.Minute)) + router.close(w) + b.Heard(w) + + got := fake.last() + if len(got.to) != 1 || got.to[0] != "ops@example.org" { + t.Fatalf("RCPT TO got %v, want the parsed address alone", got.to) + } + if !strings.Contains(got.data, `To: "Ops Team" `) { + t.Fatalf("the To header is not the formatted address:\n%s", got.data) + } +} + +// (10) A send whose ask was never published (so the router never recorded it) is settled as ended, with its own +// words, once its asked time plus SendAskLasts has passed — never left "waiting" forever (novox/hq issue 811). +func TestASendWhoseAskNeverPublishedEndsAfterTheAskWouldHaveLasted(t *testing.T) { + b, router, sends := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"}) + router.publishErr = errors.New("the bus is down") + rt := startBundle(t, b.Tools()) + + _, isErr := rt.call(SendTool, "novox.triage", sendArgs) + if !isErr { + t.Fatalf("mail_send did not report the publish failure") + } + + // The send was kept despite the publish failure (so a later sweep can still find it). + keys, err := sends.Keys() + if err != nil { + t.Fatalf("%v", err) + } + var id string + for _, k := range keys { + if strings.HasPrefix(k, "send-") { + id = k + } + } + if id == "" { + t.Fatalf("no send was kept despite the publish failure: %v", keys) + } + + // Before the ask would have lasted, it still reads waiting: the router may yet come up and record it. + text, isErr := rt.call(SendStatusTool, "novox.triage", map[string]any{"id": id}) + if isErr || !strings.Contains(text, `"status":"waiting"`) { + t.Fatalf("before expiry: %v %s", isErr, text) + } + + // Once a sweep runs after the ask would have lasted, it settles as ended, and nothing was sent. + b.Sweep(judgedAt.Add(SendAskLasts + time.Minute)) + + text, isErr = rt.call(SendStatusTool, "novox.triage", map[string]any{"id": id}) + if isErr { + t.Fatalf("%s", text) + } + if !strings.Contains(text, `"status":"ended"`) { + t.Fatalf("does not read ended once the ask would have lasted: %s", text) + } + if !strings.Contains(text, "its ask to the operator was never heard, so it was not sent") { + t.Fatalf("does not say why: %s", text) + } +} + +// (11) The sweep reads only the sends still undecided, by the undecided index, not every send ever kept: a send +// already decided is never asked about again (novox/hq issue 811). +func TestSweepReadsOnlyUndecidedSends(t *testing.T) { + fake := newFakeSMTPServer(t) + b, router, sends := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"}) + b.Dial = fake.dialer() + rt := startBundle(t, b.Tools()) + + // One send, declined right away: decided, so it leaves the undecided index. + text, isErr := rt.call(SendTool, "novox.triage", sendArgs) + if isErr { + t.Fatalf("%s", text) + } + var out map[string]any + _ = json.Unmarshal([]byte(text), &out) + declinedID := out["ask"].(string) + w := router.chose(declinedID, OptionDecline, judgedAt.Add(time.Minute)) + router.close(w) + if _, err := Decide(sends, b.lookupAccount, b.sendActor(), w, judgedAt.Add(2*time.Minute)); err != nil { + t.Fatalf("deciding: %v", err) + } + + // A second send, left undecided: still waiting on the operator. + text, isErr = rt.call(SendTool, "novox.triage", sendArgs) + if isErr { + t.Fatalf("%s", text) + } + _ = json.Unmarshal([]byte(text), &out) + waitingID := out["ask"].(string) + + router.resetRecordCalls() + b.Sweep(judgedAt.Add(3 * time.Minute)) + + if n := router.recordCallsFor(declinedID); n != 0 { + t.Fatalf("the sweep read the decided send's router record %d time(s): it already left the undecided index", n) + } + if n := router.recordCallsFor(waitingID); n == 0 { + t.Fatalf("the sweep never read the undecided send's router record") + } +} + +// (12) When the undecided index cannot be written, mail_send refuses the send and publishes no ask: nothing was +// published yet, so refusing costs nothing, and an ask the sweep could never see is never sent (novox/hq issue +// 811, review round 3). +func TestMailSendRefusesWhenTheUndecidedIndexCannotBeWritten(t *testing.T) { + b, router, sends := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"}) + sends.failKey, sends.failErr = undecidedIndexKey, errors.New("the store is down") + rt := startBundle(t, b.Tools()) + + text, isErr := rt.call(SendTool, "novox.triage", sendArgs) + if !isErr { + t.Fatalf("mail_send did not report the undecided index failing to write: %s", text) + } + if n := router.published1(); n != 0 { + t.Fatalf("an ask was published despite the undecided index failing to write: %d", n) + } +} + +// (13) An undecided index that cannot be parsed is never overwritten: addUndecided and removeUndecided fail, +// naming the key, and leave the stored bytes as they were, instead of resetting the index to empty (novox/hq +// issue 811, review round 3). +func TestAddAndRemoveUndecidedFailOnACorruptIndexRatherThanResettingIt(t *testing.T) { + sends := newMemSends() + sends.vals[undecidedIndexKey] = json.RawMessage(`not json`) + sends.revs[undecidedIndexKey] = 1 + + if err := addUndecided(sends, "send-x"); err == nil || !strings.Contains(err.Error(), undecidedIndexKey) { + t.Fatalf("addUndecided did not fail naming the key on a corrupt index: %v", err) + } + if string(sends.vals[undecidedIndexKey]) != "not json" { + t.Fatalf("addUndecided overwrote the corrupt index instead of leaving it: %s", sends.vals[undecidedIndexKey]) + } + + if err := removeUndecided(sends, "send-x"); err == nil || !strings.Contains(err.Error(), undecidedIndexKey) { + t.Fatalf("removeUndecided did not fail naming the key on a corrupt index: %v", err) + } + if string(sends.vals[undecidedIndexKey]) != "not json" { + t.Fatalf("removeUndecided overwrote the corrupt index instead of leaving it: %s", sends.vals[undecidedIndexKey]) + } +} diff --git a/modules/mesh-mail/cmd/mesh-mail/smtp.go b/modules/mesh-mail/cmd/mesh-mail/smtp.go new file mode 100644 index 00000000..bc12d406 --- /dev/null +++ b/modules/mesh-mail/cmd/mesh-mail/smtp.go @@ -0,0 +1,139 @@ +package main + +// How mail_send sends, once the operator approved it (novox/hq ADR 0361 §1): through the account's outgoing +// server, implicit TLS from the first byte on 465 or upgraded with STARTTLS on 587, as the account's outgoing +// `security` says — the same two the module already reads for an account's incoming server. The password is the +// own secret mail-credential-, the one IMAP already signs in with; it is read once, used once, and never +// logged, and neither is the body. + +import ( + "crypto/tls" + "fmt" + "mime" + "net/mail" + "net/smtp" + "strings" + "time" + "unicode" +) + +// SMTPDialer opens a connection to an outgoing server, signed in over nothing: implicit TLS on the first byte, or +// plain and then upgraded with STARTTLS, as the server's security says. A test gives a fake dialing a server of its +// own; DialSMTP is the mesh's. +type SMTPDialer func(server Server) (*smtp.Client, error) + +// DialSMTP is the dialer for the mesh (novox/hq ADR 0361 §1). +func DialSMTP(server Server) (*smtp.Client, error) { + switch server.Security { + case "tls": + conn, err := tls.Dial("tcp", server.Address(), &tls.Config{ServerName: server.Host, MinVersion: tls.VersionTLS12}) + if err != nil { + return nil, fmt.Errorf("the outgoing server %s cannot be reached over TLS: %w", server.Address(), err) + } + c, err := smtp.NewClient(conn, server.Host) + if err != nil { + _ = conn.Close() + return nil, fmt.Errorf("the outgoing server %s did not answer as SMTP: %w", server.Address(), err) + } + return c, nil + case "starttls": + c, err := smtp.Dial(server.Address()) + if err != nil { + return nil, fmt.Errorf("the outgoing server %s cannot be reached: %w", server.Address(), err) + } + if ok, _ := c.Extension("STARTTLS"); !ok { + _ = c.Close() + return nil, fmt.Errorf("the outgoing server %s does not offer STARTTLS", server.Address()) + } + if err := c.StartTLS(&tls.Config{ServerName: server.Host, MinVersion: tls.VersionTLS12}); err != nil { + _ = c.Close() + return nil, fmt.Errorf("STARTTLS to %s failed: %w", server.Address(), err) + } + return c, nil + default: + return nil, fmt.Errorf("the outgoing security %q sends over neither tls nor starttls", server.Security) + } +} + +// SendMail sends one message through the account's outgoing server (novox/hq ADR 0361 §1, §2): signs in with the +// password given, states every recipient, and writes the message once. Never logs the password or the body; an +// error names the server and what it said, never either. +func SendMail(dial SMTPDialer, a Account, password string, to []string, subject, body string, now time.Time) error { + if a.Outgoing == nil { + return fmt.Errorf("the account %s has no outgoing server", a.Name) + } + // Rcpt and the To header take the parsed address, never the raw string a caller gave: the address alone for + // Rcpt, the formatted address for the header (novox/hq issue 790, review round 1; no header injection). + rcpts := make([]string, 0, len(to)) + headers := make([]string, 0, len(to)) + for _, raw := range to { + addr, err := mail.ParseAddress(raw) + if err != nil { + return fmt.Errorf("the recipient %q is not a mail address: %w", raw, err) + } + rcpts = append(rcpts, addr.Address) + headers = append(headers, addr.String()) + } + client, err := dial(*a.Outgoing) + if err != nil { + return err + } + defer client.Close() + from := a.SignInAs() + auth := smtp.PlainAuth("", from, password, a.Outgoing.Host) + if err := client.Auth(auth); err != nil { + return fmt.Errorf("signing in to %s failed: %w", a.Outgoing.Address(), err) + } + if err := client.Mail(from); err != nil { + return fmt.Errorf("the outgoing server refused the sender: %w", err) + } + for _, rcpt := range rcpts { + if err := client.Rcpt(rcpt); err != nil { + return fmt.Errorf("the outgoing server refused the recipient %s: %w", rcpt, err) + } + } + w, err := client.Data() + if err != nil { + return fmt.Errorf("the outgoing server refused the message: %w", err) + } + if _, err := w.Write(composeMessage(from, headers, subject, body, now)); err != nil { + return fmt.Errorf("the message could not be written: %w", err) + } + if err := w.Close(); err != nil { + return fmt.Errorf("the outgoing server did not accept the message: %w", err) + } + return client.Quit() +} + +// composeMessage is a plain-text message, its lines CRLF-terminated as SMTP takes them. to is the recipients +// already formatted (mail.Address.String()), the same ones Rcpt was given, never the raw strings a caller gave. +func composeMessage(from string, to []string, subject, body string, now time.Time) []byte { + var b strings.Builder + header := func(k, v string) { b.WriteString(k + ": " + v + "\r\n") } + header("From", from) + header("To", strings.Join(to, ", ")) + header("Subject", encodeHeader(subject)) + header("Date", now.Format(time.RFC1123Z)) + header("MIME-Version", "1.0") + header("Content-Type", "text/plain; charset=utf-8") + header("Content-Transfer-Encoding", "8bit") + b.WriteString("\r\n") + b.WriteString(crlf(body)) + return []byte(b.String()) +} + +// encodeHeader word-encodes a header's value when it holds more than ASCII; a header a phone never needs to show. +func encodeHeader(s string) string { + for _, r := range s { + if r > unicode.MaxASCII { + return mime.QEncoding.Encode("UTF-8", s) + } + } + return s +} + +// crlf gives every line break in s a CR, once, the ending SMTP's DATA takes. +func crlf(s string) string { + s = strings.ReplaceAll(s, "\r\n", "\n") + return strings.ReplaceAll(s, "\n", "\r\n") +} diff --git a/modules/mesh-mail/cmd/mesh-mail/tools.go b/modules/mesh-mail/cmd/mesh-mail/tools.go index f1d57a09..3f99c806 100644 --- a/modules/mesh-mail/cmd/mesh-mail/tools.go +++ b/modules/mesh-mail/cmd/mesh-mail/tools.go @@ -1,8 +1,10 @@ package main -// The module's tools (novox/hq ADR 0359 §5): mail_accounts, mail_search, mail_read and mail_health. Who calls is the -// runtime's word (stdio.CallerOf), never an argument; a caller reads an account's mail — its search, a body, an -// attachment — only when the trusted setting `grants` names it for that account (§4). Nothing here writes to a mailbox. +// The module's tools (novox/hq ADR 0359 §5, ADR 0361 §2): mail_accounts, mail_search, mail_read, mail_health, +// mail_send and mail_send_status. Who calls is the runtime's word (stdio.CallerOf), never an argument; a caller +// reads an account's mail, or sends from it, only when the trusted setting `grants` names it for that account (§4). +// Nothing here writes to a mailbox; mail_send writes nothing itself either (send.go) — it only ever reaches the +// account's outgoing server on the operator's own Approve. import ( "context" @@ -17,9 +19,11 @@ import ( // Tool names, as the manifest declares them. const ( - AccountsTool = "mail_accounts" - SearchTool = "mail_search" - ReadTool = "mail_read" + AccountsTool = "mail_accounts" + SearchTool = "mail_search" + ReadTool = "mail_read" + SendTool = "mail_send" + SendStatusTool = "mail_send_status" ) // Bounds of what a tool answers. @@ -30,13 +34,21 @@ const ( MostAttachment = 8 << 20 ) -// Bundle is what the tools reach: the settings, the index and the attachment store. +// Bundle is what the tools reach: the settings, the index, the attachment store, and — for mail_send — the +// operator channel, the `sends` state and the outgoing dialer. The last three are nil outside a test: the real +// bus and dialer (sdkAskBus, keptSends, DialSMTP) are used lazily, so building a Bundle for the read-only tools +// never needs them. type Bundle struct { Env Env Index func(ctx context.Context) (Store, error) Objects func() (Objects, error) Caller func(args map[string]any) string Now func() time.Time + + Asks AskBus + Sends SendStore + Dial SMTPDialer + Logf func(format string, a ...any) } func (b *Bundle) now() time.Time { @@ -120,6 +132,27 @@ func (b *Bundle) Tools() []stdio.Tool { "waits, naming the account's password, while the operator has not given it at the desk.", Input: schema(nil, map[string]any{}), Run: func(map[string]any) (any, error) { return ReadStatus(b.Env.Status, b.now()), nil }}, + {Name: SendTool, + Description: "Send mail from an account you are granted (the same check as mail_read, novox/hq ADR 0361 §2). It " + + "sends nothing itself: it raises an ask on the operator channel — the account, the recipients, the subject and " + + "the first lines of the body — and answers at once with the ask's id. The mail reaches the account's outgoing " + + "server only once the operator's Approve comes back on a channel that proves who answered, and once; on " + + "Decline, or no answer before it expires, nothing is sent. Refused for an account you are not granted, or one " + + "with no outgoing server. mail_send_status, with this id, says where it stands.", + Input: schema([]string{"account", "to", "subject", "body"}, map[string]any{ + "account": text("the account to send from, by its name in mail_accounts"), + "to": map[string]any{"type": "array", "items": map[string]any{"type": "string"}, + "description": "the recipients' mail addresses, at most 20"}, + "subject": text("the message's subject, at most 200 characters"), + "body": text("the message's plain-text body"), + }), + Run: b.sendTool}, + {Name: SendStatusTool, + Description: "Where a send asked with mail_send stands, by its ask id: waiting for the operator, sent, declined, " + + "ended without a choice (expired, refused, cancelled), or not proven. Read from the router's own record, held " + + "to the request mail_send asked (so a warrant for another request, or one changed since, sends nothing here).", + Input: schema([]string{"id"}, map[string]any{"id": text("the ask id mail_send gave")}), + Run: b.sendStatusTool}, } } @@ -304,6 +337,112 @@ func refused(caller, account string) error { return fmt.Errorf("refused: %s is not granted the mail account %s; the operator grants it through the setting grants", caller, account) } +// ---- mail_send, mail_send_status --------------------------------------------------------------------------------- + +// askBus is the operator channel this bundle reaches: the bus's own, unless a test gave one. +func (b *Bundle) askBus() AskBus { + if b.Asks != nil { + return b.Asks + } + return sdkAskBus{} +} + +// sends is the `sends` state this bundle keeps: the bus's own, unless a test gave one. +func (b *Bundle) sends() SendStore { + if b.Sends != nil { + return b.Sends + } + return keptSends{stdio.State(SendsState)} +} + +// lookupAccount resolves an account by name as the settings currently hold it — read fresh at decide time, never +// the account as it stood when it was asked. +func (b *Bundle) lookupAccount(name string) (Account, bool) { + accounts, err := ReadAccounts(b.Env.Accounts) + if err != nil { + return Account{}, false + } + return accountByName(accounts, name) +} + +func accountByName(accounts []Account, name string) (Account, bool) { + for _, a := range accounts { + if a.Name == name { + return a, true + } + } + return Account{}, false +} + +// sendActor performs one send on the operator's Approve: it reads the account's own secret itself, so Decide and +// the `sends` state never hold a password. +func (b *Bundle) sendActor() SendActor { + dial := b.Dial + if dial == nil { + dial = DialSMTP + } + return func(a Account, to []string, subject, body string, now time.Time) error { + pw, given, err := ReadSecret(b.Env.Secrets, a) + if err != nil { + return fmt.Errorf("the account %s's password cannot be read: %w", a.Name, err) + } + if !given { + return fmt.Errorf("the account %s has no password yet; the operator gives it at the desk", a.Name) + } + return SendMail(dial, a, pw, to, subject, body, now) + } +} + +func stringList(v any) []string { + list, _ := v.([]any) + out := make([]string, 0, len(list)) + for _, item := range list { + if s, ok := item.(string); ok { + out = append(out, s) + } + } + return out +} + +func (b *Bundle) sendTool(args map[string]any) (any, error) { + if err := only(args, "account", "to", "subject", "body"); err != nil { + return nil, err + } + account, _ := args["account"].(string) + subject, _ := args["subject"].(string) + body, _ := args["body"].(string) + to := stringList(args["to"]) + if account == "" { + return nil, fmt.Errorf("%s needs account, by its name in mail_accounts", SendTool) + } + accounts, grants, err := b.settings() + if err != nil { + return nil, err + } + caller := b.caller(args) + // The same grants check as mail_read (novox/hq ADR 0359 §4, ADR 0361 §2): an ungranted caller is refused, and + // nothing is asked. + if !grants.Allows(caller, account) { + return nil, grants.refuse(caller, account) + } + acc, found := accountByName(accounts, account) + if !found { + return nil, fmt.Errorf("no account %q is held; mail_accounts lists the accounts", account) + } + return AskMailSend(SendInput{Account: acc, To: to, Subject: subject, Body: body, Caller: caller}, b.sends(), b.askBus(), b.now()) +} + +func (b *Bundle) sendStatusTool(args map[string]any) (any, error) { + if err := only(args, "id"); err != nil { + return nil, err + } + id, _ := args["id"].(string) + if id == "" { + return nil, fmt.Errorf("%s needs id, the ask %s gave", SendStatusTool, SendTool) + } + return ReadSendStatus(b.sends(), b.askBus(), b.lookupAccount, b.sendActor(), id, b.now()) +} + // ---- arguments ---------------------------------------------------------------------------------------------------- // only refuses an argument the tool does not take: never ignored, and none can say who calls. diff --git a/modules/mesh-mail/module.json b/modules/mesh-mail/module.json index 8a40df5d..5969c5df 100644 --- a/modules/mesh-mail/module.json +++ b/modules/mesh-mail/module.json @@ -22,14 +22,23 @@ "own-secrets": { "mail-credential-*": { "path": "${dir:state}/mail-credential-*.secret", - "issued-by": "outside" + "issued-by": "outside", + "prompt": "the password of the mail account {member}" } }, "tools": [ "mail_accounts", "mail_search", "mail_read", - "mail_health" + "mail_health", + "mail_send", + "mail_send_status" + ], + "uses": [ + "operator-channel" + ], + "state": [ + "sends" ], "streams": [ { diff --git a/modules/rofi/cmd/rofi-tools/secret.go b/modules/rofi/cmd/rofi-tools/secret.go index d6fe7164..ce95e678 100644 --- a/modules/rofi/cmd/rofi-tools/secret.go +++ b/modules/rofi/cmd/rofi-tools/secret.go @@ -46,6 +46,13 @@ var aName = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,62}$`) // break, no markup, no punctuation a sentence is written with — so the prompt's words stay this holder's. var aCaller = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._/@ -]{0,79}$`) +// escapeMarkup is `what` shown as text, never markup (novox/hq issue 789): the window's message is read with +// pango markup on, so a catalogue's words reach it with the characters pango gives meaning to turned to their +// entities first — never interpreted, whatever a module's manifest holds. +var escapeMarkupReplacer = strings.NewReplacer("&", "&", "<", "<", ">", ">") + +func escapeMarkup(s string) string { return escapeMarkupReplacer.Replace(s) } + // secretQuestion reads the call. The prompt is this holder's words, written from the module, the secret and the // machine (the review of 2026-10-09, M4): it says the mesh's controller asks, which the bus makes true — it // denies these subjects to every principal but the controller — and a caller's own words are refused, so @@ -81,14 +88,37 @@ func secretQuestion(args map[string]any) (SecretQuestion, error) { } askedBy = v } + // What the secret is for, in the catalogue's own words (novox/hq issue 789): the controller's own line, read + // from the secret's declared `prompt` and never a word the caller chose. One line, so it is shown under the + // module and secret names as its own; its characters are never markup, whatever the catalogue holds. + what := "" + if raw, there := args["what"]; there { + v, err := text(args, "what", true) + if err != nil { + return q, err + } + if strings.ContainsAny(v, "\n\r") { + return q, fmt.Errorf("what %q holds more than one line", raw) + } + what = v + } q.Prompt = names["secret"] + " for " + names["module"] q.Message = fmt.Sprintf("The mesh's controller asks for %s, the own secret of %s on %s.", names["secret"], names["module"], names["node"]) - if askedBy != "" { - q.Message += fmt.Sprintf(" It was asked for by %s.", askedBy) + if what != "" { + q.Message += "\nFor: " + escapeMarkup(what) } - q.Message += " What you type is not shown, and is sealed before it leaves this machine. Type it only if you " + + tail := "" + if askedBy != "" { + tail += fmt.Sprintf(" It was asked for by %s.", askedBy) + } + tail += " What you type is not shown, and is sealed before it leaves this machine. Type it only if you " + "expect this ask." + if what != "" { + q.Message += "\n" + strings.TrimSpace(tail) + } else { + q.Message += tail + } key, err := text(args, "seal_to", true) if err != nil { return q, err diff --git a/modules/rofi/cmd/rofi-tools/secret_test.go b/modules/rofi/cmd/rofi-tools/secret_test.go index 3aeaf2de..d9db64c6 100644 --- a/modules/rofi/cmd/rofi-tools/secret_test.go +++ b/modules/rofi/cmd/rofi-tools/secret_test.go @@ -150,3 +150,61 @@ func TestThePromptIsWrittenHereFromNamesAndSaysTheControllerAsks(t *testing.T) { } } } + +// What the secret is for shows as its own line under the module and secret names, in the catalogue's own +// words (novox/hq issue 789): read from the secret's declared `prompt`, never a word the caller chose, and +// never shown without a declared one — an old module, or one that declares none, prompts as it does today. +func TestWhatShowsAsItsOwnLineUnderTheNamesWhenDeclaredAndNothingWhenNot(t *testing.T) { + key := base64.StdEncoding.EncodeToString(make([]byte, 32)) + base := map[string]any{"module": "mesh-mail", "secret": "mail-credential-gmail", "node": "anchor", "seal_to": key} + q, err := secretQuestion(base) + if err != nil { + t.Fatal(err) + } + if strings.Contains(q.Message, "For:") { + t.Errorf("a \"For:\" line with no what declared: %q", q.Message) + } + + withWhat := map[string]any{} + for k, v := range base { + withWhat[k] = v + } + withWhat["what"] = "the password of the mail account gmail" + q, err = secretQuestion(withWhat) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(q.Message, "\nFor: the password of the mail account gmail") { + t.Errorf("no \"For:\" line of its own: %q", q.Message) + } + if !strings.Contains(q.Message, "What you type is not shown") { + t.Errorf("the rest of the message was lost: %q", q.Message) + } +} + +// What the secret is for is shown as text, never markup (novox/hq issue 789): the prompt's window reads its +// message with markup on, so what the catalogue holds reaches it with the characters markup gives meaning to +// turned to their entities first, and a line holding more than one is refused before anybody sees it. +func TestWhatIsShownAsTextNeverMarkupAndIsOneLine(t *testing.T) { + key := base64.StdEncoding.EncodeToString(make([]byte, 32)) + base := map[string]any{"module": "mesh-mail", "secret": "mail-credential-gmail", "node": "anchor", "seal_to": key, + "what": "the password of the mail & account"} + q, err := secretQuestion(base) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(q.Message, "For: the password of <b>the</b> mail & account") { + t.Errorf("markup was not escaped: %q", q.Message) + } + + for _, bad := range []string{"a\nb", "a\rb"} { + multi := map[string]any{} + for k, v := range base { + multi[k] = v + } + multi["what"] = bad + if _, err := secretQuestion(multi); err == nil { + t.Errorf("what=%q holding more than one line was taken", bad) + } + } +}