Keep every upgrade as a run on the bus and read the log's past upgrades once, so pacman_upgrades answers every node's history without asking a machine (issue 1841/S2)

This commit is contained in:
2026-10-22 14:29:23 +02:00
parent 802c72e4ed
commit efac1b4856
9 changed files with 925 additions and 0 deletions
@@ -0,0 +1,153 @@
package main
import (
"encoding/json"
"fmt"
"sort"
"strings"
"sync"
stdio "git.novox.be/novox/mesh-sdk/go"
)
// memBucket is a state in memory, with the bus's revisions and its refusal of a lost race.
type memBucket struct {
mu sync.Mutex
values map[string][]byte
revs map[string]uint64
rev uint64
writes []string // "op key", in order
fail map[string]error
}
func newBucket() *memBucket {
return &memBucket{values: map[string][]byte{}, revs: map[string]uint64{}, fail: map[string]error{}}
}
func (b *memBucket) Get(key string) (*stdio.StateEntry, error) {
b.mu.Lock()
defer b.mu.Unlock()
if err := b.fail["get "+key]; err != nil {
return nil, err
}
v, ok := b.values[key]
if !ok {
return nil, nil
}
return &stdio.StateEntry{Key: key, Value: json.RawMessage(v), Revision: b.revs[key]}, nil
}
func (b *memBucket) write(op, key string, value any) (uint64, error) {
if err := b.fail[op+" "+key]; err != nil {
return 0, err
}
data, err := json.Marshal(value)
if err != nil {
return 0, err
}
b.rev++
b.values[key], b.revs[key] = data, b.rev
b.writes = append(b.writes, op+" "+key)
return b.rev, nil
}
func (b *memBucket) Create(key string, value any) (uint64, error) {
b.mu.Lock()
defer b.mu.Unlock()
if _, ok := b.values[key]; ok {
return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
}
return b.write("create", key, value)
}
func (b *memBucket) Update(key string, value any, revision uint64) (uint64, error) {
b.mu.Lock()
defer b.mu.Unlock()
if b.revs[key] != revision {
return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
}
return b.write("update", key, value)
}
func (b *memBucket) Put(key string, value any) (uint64, error) {
b.mu.Lock()
defer b.mu.Unlock()
return b.write("put", key, value)
}
func (b *memBucket) Delete(key string) error {
b.mu.Lock()
defer b.mu.Unlock()
if err := b.fail["delete "+key]; err != nil {
return err
}
delete(b.values, key)
delete(b.revs, key)
b.writes = append(b.writes, "delete "+key)
return nil
}
func (b *memBucket) Keys() ([]string, error) {
b.mu.Lock()
defer b.mu.Unlock()
keys := []string{}
for k := range b.values {
keys = append(keys, k)
}
sort.Strings(keys)
return keys, nil
}
// value reads a key into its shape; a missing key is the zero value and false.
func (b *memBucket) value(key string, v any) bool {
b.mu.Lock()
defer b.mu.Unlock()
raw, ok := b.values[key]
if ok {
if err := json.Unmarshal(raw, v); err != nil {
panic(err)
}
}
return ok
}
// writesTo are the writes to keys starting with prefix.
func (b *memBucket) writesTo(prefix string) []string {
b.mu.Lock()
defer b.mu.Unlock()
var out []string
for _, w := range b.writes {
if _, key, _ := strings.Cut(w, " "); strings.HasPrefix(key, prefix) {
out = append(out, w)
}
}
return out
}
// sentEvent is one event a fake sink was handed.
type sentEvent struct {
Name string
Body any
}
type eventLog struct {
mu sync.Mutex
sent []sentEvent
}
func (e *eventLog) Emit(name string, body any, _ ...stdio.EmitOptions) error {
e.mu.Lock()
defer e.mu.Unlock()
e.sent = append(e.sent, sentEvent{name, body})
return nil
}
func (e *eventLog) names() []string {
e.mu.Lock()
defer e.mu.Unlock()
var out []string
for _, s := range e.sent {
out = append(out, s.Name)
}
return out
}
+28
View File
@@ -159,6 +159,34 @@ func tools(m *Machine) []stdio.Tool {
Input: schema(map[string]any{}),
Run: func(map[string]any) (any, error) { return m.UpgradeHealth(), nil },
},
{
Name: "pacman_upgrades",
Description: "Every full upgrade and rollback of every node, from the bus (ADR 0508 §1): each run's node, kind, what started it, " +
"its round, when it began and ended, its outcome (refused, finished, failed or running) and whether its after-upgrade check passed; " +
"narrowed to a node and to days, newest first. With a node and a run, that run whole, every step with its time.",
Input: schema(map[string]any{
"node": map[string]any{"type": "string", "description": "only this node's runs"},
"from": sinceArg,
"until": map[string]any{"type": "string", "description": "a day, YYYY-MM-DD, the last one listed"},
"run": map[string]any{"type": "string", "description": "one run, by its id, with node"},
"limit": limitArg(50, 1000),
}),
Run: func(args map[string]any) (any, error) {
var v [4]string
for i, k := range []string{"node", "from", "until", "run"} {
s, err := text(args, k, false)
if err != nil {
return nil, err
}
v[i] = s
}
n, err := whole(args, "limit", 50, 1, 1000)
if err != nil {
return nil, err
}
return ListUpgrades(upgradeState(StateRuns), v[0], v[1], v[2], v[3], n)
},
},
{
Name: "pacman_orphans",
Description: "Packages installed as dependencies that nothing requires any more (pacman -Qdt), with versions.",
@@ -0,0 +1,217 @@
package main
// Runs on the bus and the log read once (novox/hq ADR 0508 §1; acceptance 1841/AC1 for the writing of steps,
// 1841/AC2 for the log and pacman_upgrades).
import (
"errors"
"os"
"strings"
"testing"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
)
func capturedLog(t *testing.T) string {
t.Helper()
raw, err := os.ReadFile("testdata/pacman-log-three-full-upgrades/pacman.log")
if err != nil {
t.Fatal(err)
}
return string(raw)
}
// The captured log holds three full upgrades among other commands: one that ended before any transaction, one
// whose transaction completed, and a later one that upgraded one package.
func TestEveryFullUpgradeOfTheLogIsARunReadFromIt(t *testing.T) {
runs := PastUpgrades("node-a", capturedLog(t))
if len(runs) != 3 {
t.Fatalf("%d runs, want the three full upgrades and none of the other commands: %+v", len(runs), runs)
}
for _, r := range runs {
if r.Node != "node-a" || r.Kind != RunUpgrade || r.From != FromLog || len(r.Steps) != 2 || r.Steps[0].Step != StepStarted {
t.Errorf("a run read from the log: %+v", r)
}
}
first, second, third := runs[0], runs[1], runs[2]
if first.Run != "20261004T130648Z" || first.Steps[1].Step != StepFinished || first.Steps[1].Why != noTransaction ||
len(first.Steps[1].Changes.Changed) != 0 {
t.Errorf("an upgrade with no transaction says so: %+v", first)
}
end := second.Steps[1]
if second.Run != "20261004T130724Z" || end.Step != StepFinished || !end.At.Equal(time.Date(2026, 10, 4, 13, 22, 50, 0, time.UTC)) {
t.Errorf("the second upgrade finished when its transaction completed: %+v", second)
}
var names []string
for _, c := range end.Changes.Changed {
names = append(names, c.Name)
}
if strings.Join(names, " ") != "bash glibc iana-etc readline tzdata wireshark-qt xfsprogs xpad xterm zathura-pdf-mupdf" {
t.Errorf("what it changed, sorted: %v", names)
}
if c := end.Changes.Changed[0]; c.From != "5.3.15-1" || c.To != "5.3.20-1" {
t.Errorf("bash: %+v", c)
}
if len(end.Changes.Removed) != 0 {
t.Errorf("the removal run between them is no part of an upgrade: %+v", end.Changes.Removed)
}
if c := third.Steps[1].Changes.Changed; third.Run != "20261004T132817Z" || len(c) != 1 || c[0].Name != "dunst" {
t.Errorf("the third upgrade: %+v", third)
}
}
func TestATransactionWithoutAnEndOrThatFailedIsAFailedRun(t *testing.T) {
log := "[2026-10-10T04:00:00+0000] [PACMAN] Running 'pacman -Syu --noconfirm'\n" +
"[2026-10-10T04:00:01+0000] [PACMAN] starting full system upgrade\n" +
"[2026-10-10T04:00:05+0000] [ALPM] transaction started\n" +
"[2026-10-10T04:00:06+0000] [ALPM] upgraded linux (7.2.8.arch1-2 -> 7.2.9.arch1-1)\n" +
"[2026-10-10T04:00:09+0000] [ALPM] transaction failed\n" +
"[2026-10-11T04:00:00+0000] [PACMAN] Running 'pacman -Syu --noconfirm'\n" +
"[2026-10-11T04:00:01+0000] [PACMAN] starting full system upgrade\n" +
"[2026-10-11T04:00:05+0000] [ALPM] transaction started\n"
runs := PastUpgrades("node-b", log)
if len(runs) != 2 {
t.Fatalf("%+v", runs)
}
if s := runs[0].Steps[1]; s.Step != StepFailed || s.Error != "transaction failed" || len(s.Changes.Changed) != 1 {
t.Errorf("a failed transaction: %+v", s)
}
if s := runs[1].Steps[1]; s.Step != StepFailed || !strings.Contains(s.Error, "no end") {
t.Errorf("a transaction the log holds no end of: %+v", s)
}
}
func TestTheLogIsReadIntoRunsOnceAndASecondStartReadsNothing(t *testing.T) {
m := machine(nil, 1000)
reads := 0
log := capturedLog(t)
m.ReadFile = func(path string) ([]byte, error) {
if path != PacmanLog {
t.Fatalf("read %s", path)
}
reads++
return []byte(log), nil
}
runs := newBucket()
other := Run{Node: "node-b", Run: "20261001T040000Z", Kind: RunUpgrade, From: FromRound}
if _, err := runs.Create(NodeKey("node-b", other.Run), other); err != nil {
t.Fatal(err)
}
u := &Upgrades{Node: "node-a", Machine: m, Runs: runs}
n, err := u.ImportLog()
if err != nil || n != 3 || reads != 1 {
t.Fatalf("first start: %d runs, %d reads, %v", n, reads, err)
}
var r Run
if !runs.value("node-a.20261004T132817Z", &r) || r.From != FromLog {
t.Errorf("the run is kept at <node>.<run>, marked as read from the log: %+v", r)
}
before := len(runs.writes)
n, err = u.ImportLog()
if err != nil || n != 0 || reads != 1 || len(runs.writes) != before {
t.Errorf("a second start: %d runs, %d reads, %d writes, %v", n, reads, len(runs.writes)-before, err)
}
}
// A run's key is created at its first step and then only updated from the revision its writer holds: a step
// is written once, and a run another writer changed meanwhile is refused, never overwritten.
func TestAStepIsWrittenOnceAndAnotherWritersChangeIsNeverOverwritten(t *testing.T) {
runs := newBucket()
u := &Upgrades{Node: "node-a", Runs: runs}
r := &openRun{Run: Run{Node: "node-a", Run: "20261024T040000Z", Kind: RunUpgrade, From: FromAsk}}
at := time.Date(2026, 10, 24, 4, 0, 0, 0, time.UTC)
for _, s := range []string{StepRecorded, StepCheckedBefore} {
if err := u.step(r, Step{Step: s, At: at}); err != nil {
t.Fatal(err)
}
}
if w := strings.Join(runs.writes, ", "); w != "create node-a.20261024T040000Z, update node-a.20261024T040000Z" {
t.Errorf("writes: %s", w)
}
var kept Run
runs.value("node-a.20261024T040000Z", &kept)
if len(kept.Steps) != 2 || kept.Steps[1].Step != StepCheckedBefore {
t.Errorf("kept: %+v", kept)
}
if _, err := runs.Put("node-a.20261024T040000Z", kept); err != nil {
t.Fatal(err)
}
if err := u.step(r, Step{Step: StepStarted, At: at}); !errors.Is(err, stdio.ErrChanged) || len(r.Steps) != 2 {
t.Errorf("a step over another writer's change: %v, %d steps held", err, len(r.Steps))
}
}
func TestPacmanUpgradesAnswersAllOneNodeDaysAndOneRun(t *testing.T) {
runs := newBucket()
put := func(node, id string, steps ...Step) {
r := Run{Node: node, Run: id, Kind: RunUpgrade, From: FromRound, Round: "2026-W43", Steps: steps}
if _, err := runs.Create(NodeKey(node, id), r); err != nil {
t.Fatal(err)
}
}
at := time.Date(2026, 10, 24, 4, 30, 0, 0, time.UTC)
passed := &Check{Passed: true}
put("node-a", "20261024T040000Z", Step{Step: StepRecorded, At: at}, Step{Step: StepFinished, At: at}, Step{Step: StepCheckedAfter, At: at, Check: passed})
put("node-b", "20261024T050000Z", Step{Step: StepRecorded, At: at}, Step{Step: StepStarted, At: at})
put("node-b", "20261017T050000Z", Step{Step: StepRefused, At: at, Why: "installed is not current"})
all, err := ListUpgrades(runs, "", "", "", "", 50)
if err != nil {
t.Fatal(err)
}
listed := all["runs"].([]RunSummary)
var order []string
for _, s := range listed {
order = append(order, s.Node+"/"+s.Run+"/"+s.Outcome+"/"+s.Checked)
}
if strings.Join(order, " ") != "node-b/20261024T050000Z/running/ node-a/20261024T040000Z/finished/passed node-b/20261017T050000Z/refused/" {
t.Fatalf("all runs, newest first: %v", order)
}
if listed[2].Error != "installed is not current" {
t.Errorf("a refused run says why: %+v", listed[2])
}
one, _ := ListUpgrades(runs, "node-b", "", "", "", 50)
if one["count"] != 2 {
t.Errorf("one node: %v", one)
}
days, _ := ListUpgrades(runs, "", "2026-10-18", "2026-10-24", "", 50)
if days["count"] != 2 {
t.Errorf("between two days, the last one included: %v", days)
}
limited, _ := ListUpgrades(runs, "", "", "", "", 1)
if limited["count"] != 1 || limited["truncated"] != true {
t.Errorf("limited: %v", limited)
}
whole, err := ListUpgrades(runs, "node-a", "", "", "20261024T040000Z", 50)
if err != nil || len(whole["run"].(Run).Steps) != 3 {
t.Errorf("one run whole: %v %v", whole, err)
}
if _, err := ListUpgrades(runs, "node-a", "", "", "20261024T050000Z", 50); err == nil {
t.Error("a run the node does not have was answered")
}
if _, err := ListUpgrades(runs, "", "", "", "20261024T040000Z", 50); err == nil {
t.Error("a run without its node was answered")
}
}
func TestTheUpgradesToolReadsTheRunsState(t *testing.T) {
runs := newBucket()
saved := upgradeState
upgradeState = func(name string) Bucket {
if name != StateRuns {
t.Fatalf("read %s", name)
}
return runs
}
defer func() { upgradeState = saved }()
for _, tool := range tools(machine(nil, 1000)) {
if tool.Name == "pacman_upgrades" {
out, err := tool.Run(map[string]any{"node": "node-a"})
if err != nil || out.(map[string]any)["count"] != 0 {
t.Errorf("%v %v", out, err)
}
return
}
}
t.Fatal("no pacman_upgrades")
}
@@ -0,0 +1,4 @@
openrazer-driver/3.12.4, 7.2.8-arch1-2, x86_64: installed
vboxhost/7.2.20_OSE, 7.2.8-arch1-2, x86_64: installed
xone/0.5.8, 7.2.8-arch1-2, x86_64: installed
xpad/0.4: added
@@ -0,0 +1,10 @@
Meta: 1
Name: zfs
Branch: 1.0
Version: 2.4.4
Release: 1
Release-Tags: relext
License: CDDL
Author: OpenZFS
Linux-Maximum: 7.2
Linux-Minimum: 4.18
@@ -0,0 +1,56 @@
PACKAGE_NAME="zfs"
PACKAGE_VERSION="2.4.4"
PACKAGE_CONFIG="/etc/sysconfig/zfs"
NO_WEAK_MODULES="yes"
PRE_BUILD="configure
--disable-dependency-tracking
--prefix=/usr
--with-config=kernel
--with-linux="${kernel_source_dir}"
--with-linux-obj="${kernel_source_dir}"
$(
[[ -n \"${ICP_ROOT}\" ]] && \
{
echo --with-qat=\"${ICP_ROOT}\"
}
)
$(
[[ -r ${PACKAGE_CONFIG} ]] \
&& source ${PACKAGE_CONFIG} \
&& shopt -q -s extglob \
&& \
{
if [[ ${ZFS_DKMS_ENABLE_DEBUG,,} == @(y|yes) ]]
then
echo --enable-debug
fi
if [[ ${ZFS_DKMS_ENABLE_DEBUGINFO,,} == @(y|yes) ]]
then
echo --enable-debuginfo
fi
}
)
"
POST_BUILD="scripts/dkms.postbuild
-n ${PACKAGE_NAME}
-v ${PACKAGE_VERSION}
-a ${arch}
-k ${kernelver}
-t ${dkms_tree}
"
AUTOINSTALL="yes"
MAKE[0]="make -C module/"
STRIP[0]="$(
[[ -r ${PACKAGE_CONFIG} ]] \
&& source ${PACKAGE_CONFIG} \
&& shopt -q -s extglob \
&& [[ ${ZFS_DKMS_DISABLE_STRIP,,} == @(y|yes) ]] \
&& echo -n no
)"
STRIP[1]="${STRIP[0]}"
BUILT_MODULE_NAME[0]="zfs"
BUILT_MODULE_LOCATION[0]="module/"
DEST_MODULE_LOCATION[0]="/extra"
BUILT_MODULE_NAME[1]="spl"
BUILT_MODULE_LOCATION[1]="module/"
DEST_MODULE_LOCATION[1]="/extra"
@@ -0,0 +1,64 @@
[2026-10-04T14:52:51+0200] [ALPM] installed logrotate (3.22.0-1)
[2026-10-04T14:52:51+0200] [ALPM] transaction completed
[2026-10-04T14:52:51+0200] [ALPM] running '30-systemd-daemon-reload-system.hook'...
[2026-10-04T14:54:22+0200] [ALPM] running '35-systemd-update.hook'...
[2026-10-04T15:06:13+0200] [PACMAN] Running 'pacman -Syu --noconfirm'
[2026-10-04T15:06:13+0200] [PACMAN] synchronizing package lists
[2026-10-04T15:06:48+0200] [PACMAN] starting full system upgrade
[2026-10-04T15:07:23+0200] [PACMAN] Running 'pacman -Rns --noconfirm lib32-libpcap'
[2026-10-04T15:07:23+0200] [ALPM] transaction started
[2026-10-04T15:07:24+0200] [ALPM] removed lib32-libpcap (1.10.6-1)
[2026-10-04T15:07:24+0200] [ALPM] removed lib32-libnl (3.12.0-1)
[2026-10-04T15:07:24+0200] [ALPM] transaction completed
[2026-10-04T15:07:24+0200] [ALPM] running '35-systemd-update.hook'...
[2026-10-04T15:07:24+0200] [PACMAN] Running 'pacman -Syu --noconfirm'
[2026-10-04T15:07:24+0200] [PACMAN] synchronizing package lists
[2026-10-04T15:07:24+0200] [PACMAN] starting full system upgrade
[2026-10-04T15:16:51+0200] [ALPM] running '60-mkinitcpio-remove.hook'...
[2026-10-04T15:16:52+0200] [ALPM] running '71-dkms-remove.hook'...
[2026-10-04T15:16:52+0200] [ALPM-SCRIPTLET] ==> dkms remove --no-depmod xone/0.5.8 -k 7.1.9-arch1-2
[2026-10-04T15:16:52+0200] [ALPM-SCRIPTLET] ==> depmod 7.1.9-arch1-2
[2026-10-04T15:16:55+0200] [ALPM] running 'ghc-unregister.hook'...
[2026-10-04T15:16:57+0200] [ALPM] running 'texinfo-remove.hook'...
[2026-10-04T15:16:58+0200] [ALPM] transaction started
[2026-10-04T15:16:58+0200] [ALPM] upgraded tzdata (2026c-1 -> 2026e-1)
[2026-10-04T15:16:58+0200] [ALPM] upgraded iana-etc (20260530-1 -> 20260617-1)
[2026-10-04T15:16:58+0200] [ALPM] upgraded glibc (2.44+r24+g16be1518495f-1 -> 2.44+r50+g1848099f063e-1)
[2026-10-04T15:16:58+0200] [ALPM] upgraded readline (8.3.003-1 -> 8.3.6-1)
[2026-10-04T15:16:58+0200] [ALPM] upgraded bash (5.3.15-1 -> 5.3.20-1)
[2026-10-04T15:22:46+0200] [ALPM] upgraded wireshark-qt (4.7.2-1 -> 4.7.3-1)
[2026-10-04T15:22:46+0200] [ALPM] upgraded xfsprogs (7.1.1-1 -> 7.2.0-1)
[2026-10-04T15:22:47+0200] [ALPM] upgraded xpad (5.8.0-1 -> 5.9.0-1)
[2026-10-04T15:22:47+0200] [ALPM] upgraded xterm (410-1 -> 411-1)
[2026-10-04T15:22:47+0200] [ALPM] upgraded zathura-pdf-mupdf (2026.07.18-1 -> 2026.07.18-4)
[2026-10-04T15:22:50+0200] [ALPM] transaction completed
[2026-10-04T15:22:52+0200] [ALPM] running '10-glibc-locale-gen.hook'...
[2026-10-04T15:22:52+0200] [ALPM-SCRIPTLET] Generating locales...
[2026-10-04T15:22:53+0200] [ALPM-SCRIPTLET] en_US.UTF-8... done
[2026-10-04T15:22:53+0200] [ALPM-SCRIPTLET] Generation complete.
[2026-10-04T15:22:53+0200] [ALPM] running '11-glibc-ldconfig.hook'...
[2026-10-04T15:24:49+0200] [ALPM] running 'texinfo-install.hook'...
[2026-10-04T15:24:51+0200] [ALPM] running 'update-desktop-database.hook'...
[2026-10-04T15:24:51+0200] [ALPM] running 'update-vlc-plugin-cache.hook'...
[2026-10-04T15:24:52+0200] [ALPM] running 'xorg-mkfontscale.hook'...
[2026-10-04T15:28:15+0200] [PACMAN] Running 'pacman -Syu'
[2026-10-04T15:28:15+0200] [PACMAN] synchronizing package lists
[2026-10-04T15:28:17+0200] [PACMAN] starting full system upgrade
[2026-10-04T15:30:26+0200] [ALPM] transaction started
[2026-10-04T15:30:26+0200] [ALPM] upgraded dunst (1.13.2-2 -> 1.13.2-3)
[2026-10-04T15:30:26+0200] [ALPM] transaction completed
[2026-10-04T15:30:26+0200] [ALPM] running '30-systemd-daemon-reload-user.hook'...
[2026-10-04T15:30:26+0200] [ALPM-SCRIPTLET] Failed to connect to system scope bus via local transport: Connection refused
[2026-10-04T15:30:26+0200] [ALPM] running '35-systemd-update.hook'...
[2026-10-04T15:30:26+0200] [ALPM] running 'i3status-rust.hook'...
[2026-10-04T16:00:43+0200] [PACMAN] Running 'pacman -R --noconfirm ntp'
[2026-10-04T16:00:43+0200] [ALPM] transaction started
[2026-10-04T16:00:45+0200] [ALPM] removed ntp (4.2.8.p18-6)
[2026-10-04T16:00:45+0200] [ALPM] transaction completed
[2026-10-04T16:00:45+0200] [ALPM] running '30-systemd-daemon-reload-system.hook'...
[2026-10-04T16:00:45+0200] [ALPM] running '35-systemd-update.hook'...
[2026-10-05T10:04:50+0200] [PACMAN] Running 'pacman -S --noconfirm --ask 4 lemurs'
[2026-10-05T10:04:51+0200] [ALPM] transaction started
[2026-10-05T10:04:51+0200] [ALPM] warning: /etc/lemurs/config.toml saved as /etc/lemurs/config.toml.pacsave
[2026-10-05T10:04:51+0200] [ALPM] removed lemurs-git (0.4.0.r17.g7151336-2)
[2026-10-05T10:04:51+0200] [ALPM] installed lemurs (0.4.0-1)
+392
View File
@@ -0,0 +1,392 @@
package main
// Upgrades as runs on the bus (novox/hq ADR 0508 §1): every full upgrade of a node, and every rollback, is a run
// in the state `runs` at `<node>.<run>`, written at each of its steps and at no other time, and each step is also
// an event. The keeper's loop calls what is here; everything it reaches is behind an interface (a state, the
// events, the root acts) or the Machine's injected runner and files, so it is tested without a bus or a machine.
import (
"encoding/json"
"errors"
"fmt"
"regexp"
"sort"
"strings"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
)
// Bucket is one of the module's states as a writer sees it: a process's state, or a tools bundle's
// (stdio.KeptState). A write that lost a race is stdio.ErrChanged.
type Bucket interface {
Get(key string) (*stdio.StateEntry, error)
Create(key string, value any) (uint64, error)
Update(key string, value any, revision uint64) (uint64, error)
Put(key string, value any) (uint64, error)
Delete(key string) error
Keys() ([]string, error)
}
// EventSink emits the keeper's events: the process's connection (process.Conn).
type EventSink interface {
Emit(eventType string, body any, opts ...stdio.EmitOptions) error
}
// RootActs are the acts that need root, carried out by the module's root helper (ADR 0510): the keeper never
// runs them itself.
type RootActs interface {
// Transaction starts the run's transaction as a unit of the service manager, named TransactionUnit(kind,
// run), leaving out the packages ignore names, and answers once it started.
Transaction(kind, run string, ignore []string) error
// Journal is the last n lines a transaction unit logged.
Journal(unit string, n int) ([]string, error)
}
// TransactionUnit is the unit a run's transaction runs as: named by the run, so a keeper started again finds it.
func TransactionUnit(kind, run string) string {
return unitPrefix + kind + "-" + strings.ToLower(run)
}
// Upgrades is one node's upgrades: what the keeper's loop calls for its own node.
type Upgrades struct {
Node string
Config KeeperConfig
Machine *Machine
Root RootActs
Events EventSink
// The module's states.
Installed, Updates, Runs, Records, Asks Bucket
// Check is the after-upgrade check (ADR 0508 §4); nil checks only that the transaction succeeded.
Check func(run Run, record *Record) Check
}
// openRun is a run being written, with the revision of its key.
type openRun struct {
Run
rev uint64
}
// step writes the run with one more step: created at its first step, then only ever updated from the revision
// this writer wrote, so a second writer is refused rather than overwritten.
func (u *Upgrades) step(r *openRun, s Step) error {
r.Steps = append(r.Steps, s)
key := NodeKey(r.Node, r.Run.Run)
var err error
if r.rev == 0 {
r.rev, err = u.Runs.Create(key, r.Run)
} else {
r.rev, err = u.Runs.Update(key, r.Run, r.rev)
}
if err != nil {
r.Steps = r.Steps[:len(r.Steps)-1]
return fmt.Errorf("writing step %s of run %s: %w", s.Step, key, err)
}
return nil
}
// emit says a step as its event; an event that is not sent is said in the error, the step stays written.
func (u *Upgrades) emit(event string, body any) error {
if u.Events == nil {
return nil
}
if err := u.Events.Emit(event, body); err != nil {
return fmt.Errorf("emitting %s: %w", event, err)
}
return nil
}
// The package manager's log, read as blocks: each begins at a `Running '<command>'` line (or a full upgrade
// starting with none before it) and holds the transaction it ran and the packages it changed.
var (
runningLine = regexp.MustCompile(`^\[([^\]]+)\] \[PACMAN\] Running '(.*)'$`)
transactionLine = regexp.MustCompile(`^\[([^\]]+)\] \[ALPM\] transaction (started|completed|failed|interrupted)$`)
)
// logBlock is one command pacman ran, as its log tells it.
type logBlock struct {
Start time.Time
Command string
Full bool // it started a full system upgrade
FullAt time.Time // when
Began bool // a transaction started
End string // completed, failed or interrupted; empty when the log holds no end
EndAt time.Time
Changes Changes
}
// logBlocks reads pacman's log into the commands it ran, oldest first.
func logBlocks(text string) []logBlock {
var blocks []logBlock
var cur *logBlock
open := func(at time.Time, command string) {
blocks = append(blocks, logBlock{Start: at, Command: command})
cur = &blocks[len(blocks)-1]
}
for _, l := range strings.Split(text, "\n") {
l = strings.TrimRight(l, "\r")
if m := runningLine.FindStringSubmatch(l); m != nil {
if t, ok := logTime(m[1]); ok {
open(t, m[2])
}
continue
}
if m := fullUpdate.FindStringSubmatch(l); m != nil {
t, ok := logTime(m[1])
if !ok {
continue
}
if cur == nil || cur.Full || cur.Began {
open(t, "")
}
cur.Full, cur.FullAt = true, t
continue
}
if cur == nil {
continue
}
if m := transactionLine.FindStringSubmatch(l); m != nil {
t, _ := logTime(m[1])
if m[2] == "started" {
cur.Began = true
} else if cur.Began && cur.End == "" {
cur.End, cur.EndAt = m[2], t
}
continue
}
if cur.Began && cur.End == "" {
if m := logLine.FindStringSubmatch(l); m != nil {
addChange(&cur.Changes, m[2], m[3], m[4])
}
}
}
for i := range blocks {
sortChanges(&blocks[i].Changes)
}
return blocks
}
func addChange(c *Changes, action, name, version string) {
switch action {
case "installed":
c.Added = append(c.Added, PackageVersion{Name: name, Version: version})
case "removed":
c.Removed = append(c.Removed, PackageVersion{Name: name, Version: version})
case "upgraded", "downgraded":
from, to, _ := strings.Cut(version, " -> ")
c.Changed = append(c.Changed, VersionChange{Name: name, From: from, To: to})
}
}
func sortChanges(c *Changes) {
if c.Added == nil {
c.Added = []PackageVersion{}
}
if c.Removed == nil {
c.Removed = []PackageVersion{}
}
if c.Changed == nil {
c.Changed = []VersionChange{}
}
sort.Slice(c.Added, func(i, j int) bool { return c.Added[i].Name < c.Added[j].Name })
sort.Slice(c.Removed, func(i, j int) bool { return c.Removed[i].Name < c.Removed[j].Name })
sort.Slice(c.Changed, func(i, j int) bool { return c.Changed[i].Name < c.Changed[j].Name })
}
// noTransaction is why a full upgrade the log tells of has no transaction: the log does not say which.
const noTransaction = "the log holds no transaction for this upgrade: there was nothing to do, or it ended before changing anything"
// PastUpgrades is every full upgrade pacman's log tells of, as runs read from the log (ADR 0508 §1): started
// when the upgrade started, finished or failed when its transaction ended, with what it changed. One whose
// transaction has no end in the log failed, so said.
func PastUpgrades(node, text string) []Run {
var runs []Run
for _, b := range logBlocks(text) {
if !b.Full {
continue
}
r := Run{Node: node, Run: RunID(b.FullAt), Kind: RunUpgrade, From: FromLog,
Steps: []Step{{Step: StepStarted, At: b.FullAt}}}
changes := b.Changes
switch {
case !b.Began:
r.Steps = append(r.Steps, Step{Step: StepFinished, At: b.FullAt, Why: noTransaction, Changes: &changes})
case b.End == "completed":
r.Steps = append(r.Steps, Step{Step: StepFinished, At: b.EndAt, Changes: &changes})
case b.End == "":
r.Steps = append(r.Steps, Step{Step: StepFailed, At: b.FullAt, Error: "the log holds no end of this transaction", Changes: &changes})
default:
r.Steps = append(r.Steps, Step{Step: StepFailed, At: b.EndAt, Error: "transaction " + b.End, Changes: &changes})
}
runs = append(runs, r)
}
return runs
}
// ImportLog reads the node's past full upgrades from pacman's log into `runs` once: when the node has no run
// yet. It answers how many it wrote. A run read from the log is history and emits nothing.
func (u *Upgrades) ImportLog() (int, error) {
keys, err := u.Runs.Keys()
if err != nil {
return 0, fmt.Errorf("listing runs: %w", err)
}
for _, k := range keys {
if strings.HasPrefix(k, u.Node+".") {
return 0, nil
}
}
text, err := u.Machine.ReadFile(PacmanLog)
if err != nil {
return 0, fmt.Errorf("reading %s: %w", PacmanLog, err)
}
n := 0
for _, r := range PastUpgrades(u.Node, string(text)) {
if _, err := u.Runs.Create(NodeKey(u.Node, r.Run), r); err != nil {
if errors.Is(err, stdio.ErrChanged) {
continue // two upgrades started in the same second: the first is kept
}
return n, fmt.Errorf("writing run %s: %w", r.Run, err)
}
n++
}
return n, nil
}
// runTime is when a run began, from its id.
func runTime(id string) (time.Time, bool) {
t, err := time.Parse("20060102T150405Z", id)
return t, err == nil
}
// RunSummary is one run as pacman_upgrades lists it.
type RunSummary struct {
Node string `json:"node"`
Run string `json:"run"`
Kind string `json:"kind"`
From string `json:"from"`
Round string `json:"round,omitempty"`
Began time.Time `json:"began"`
Outcome string `json:"outcome"`
Ended *time.Time `json:"ended,omitempty"`
Error string `json:"error,omitempty"`
Held int `json:"held_kernels,omitempty"`
Checked string `json:"check,omitempty"`
}
// summary is where a run stands: its outcome is its last step that ends it (refused, finished, failed), else
// running.
func summary(r Run) RunSummary {
s := RunSummary{Node: r.Node, Run: r.Run, Kind: r.Kind, From: r.From, Round: r.Round, Outcome: "running"}
if t, ok := runTime(r.Run); ok {
s.Began = t
}
for _, st := range r.Steps {
switch st.Step {
case StepRefused, StepFinished, StepFailed:
at := st.At
s.Outcome, s.Ended = st.Step, &at
s.Error = st.Error
if st.Step == StepRefused {
s.Error = st.Why
}
case StepCheckedBefore:
s.Held = len(st.Held)
case StepCheckedAfter:
s.Checked = "failed"
if st.Check != nil && st.Check.Passed {
s.Checked = "passed"
}
}
}
return s
}
// ListUpgrades answers pacman_upgrades from `runs`: one run whole, or the runs of every node or one, between
// two days, newest first, at most limit of them.
func ListUpgrades(runs Bucket, node, from, until, run string, limit int) (map[string]any, error) {
if run != "" {
if node == "" {
return nil, fmt.Errorf("a run is named with its node")
}
e, err := runs.Get(NodeKey(node, run))
if err != nil {
return nil, err
}
if e == nil {
return nil, fmt.Errorf("no run %s of %s", run, node)
}
var r Run
if err := jsonValue(e.Value, &r); err != nil {
return nil, err
}
return map[string]any{"run": r, "summary": summary(r)}, nil
}
day := func(s string, end bool) (time.Time, error) {
if s == "" {
return time.Time{}, nil
}
t, err := time.ParseInLocation("2006-01-02", s, time.Local)
if err != nil {
return t, fmt.Errorf("%q is not a day as YYYY-MM-DD", s)
}
if end {
t = t.AddDate(0, 0, 1)
}
return t, nil
}
since, err := day(from, false)
if err != nil {
return nil, err
}
before, err := day(until, true)
if err != nil {
return nil, err
}
keys, err := runs.Keys()
if err != nil {
return nil, err
}
var listed []RunSummary
for _, k := range keys {
n, id, ok := strings.Cut(k, ".")
if !ok || node != "" && n != node {
continue
}
t, ok := runTime(id)
if !ok || t.Before(since) || !before.IsZero() && !t.Before(before) {
continue
}
e, err := runs.Get(k)
if err != nil {
return nil, err
}
if e == nil {
continue
}
var r Run
if err := jsonValue(e.Value, &r); err != nil {
return nil, fmt.Errorf("run %s: %w", k, err)
}
listed = append(listed, summary(r))
}
sort.Slice(listed, func(i, j int) bool {
if !listed[i].Began.Equal(listed[j].Began) {
return listed[i].Began.After(listed[j].Began)
}
return listed[i].Node < listed[j].Node
})
truncated := len(listed) > limit
if truncated {
listed = listed[:limit]
}
if listed == nil {
listed = []RunSummary{}
}
return map[string]any{"count": len(listed), "runs": listed, "truncated": truncated}, nil
}
// jsonValue reads a state's value into its shape.
func jsonValue(raw []byte, v any) error { return json.Unmarshal(raw, v) }
// upgradeState is a state of this module as the tools bundle reads it; a test replaces it.
var upgradeState = func(name string) Bucket { return stdio.State(name) }
+1
View File
@@ -71,6 +71,7 @@
"pacman_updates",
"pacman_upgrade",
"pacman_upgrade_health",
"pacman_upgrades",
"pacman_orphans",
"pacman_remove_orphans",
"pacman_cache",