631 lines
18 KiB
Go
631 lines
18 KiB
Go
package main
|
|
|
|
// The keeper's start and its lock watch (novox/hq ADR 0507 §2, §3; acceptance 1840/AC1-AC3), over a fake state,
|
|
// a fake machine, a fake clock and a fake watch.
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
)
|
|
|
|
// fakeStore is one state in memory, recording every write.
|
|
type fakeStore struct {
|
|
mu sync.Mutex
|
|
name string
|
|
values map[string]json.RawMessage
|
|
revs map[string]uint64
|
|
at map[string]time.Time
|
|
rev uint64
|
|
writes []string // "put <key>", "delete <key>"
|
|
watchers []fakeWatch
|
|
now func() time.Time
|
|
}
|
|
|
|
type fakeWatch struct {
|
|
key string
|
|
handle func(stdio.StateChange) error
|
|
}
|
|
|
|
func newFakeStore(name string) *fakeStore {
|
|
return &fakeStore{name: name, values: map[string]json.RawMessage{}, revs: map[string]uint64{}, at: map[string]time.Time{}}
|
|
}
|
|
|
|
func (s *fakeStore) Get(key string) (*stdio.StateEntry, error) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
v, ok := s.values[key]
|
|
if !ok {
|
|
return nil, nil
|
|
}
|
|
return &stdio.StateEntry{Key: key, Value: v, Revision: s.revs[key]}, nil
|
|
}
|
|
|
|
func (s *fakeStore) write(key string, value any, op string) (uint64, error) {
|
|
data, err := json.Marshal(value)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
s.mu.Lock()
|
|
s.rev++
|
|
s.values[key], s.revs[key], s.at[key] = data, s.rev, s.clock()
|
|
s.writes = append(s.writes, op+" "+key)
|
|
rev, ws := s.rev, append([]fakeWatch(nil), s.watchers...)
|
|
s.mu.Unlock()
|
|
for _, w := range ws {
|
|
if matchKey(w.key, key) {
|
|
w.handle(stdio.StateChange{State: s.name, Key: key, Op: "put", Value: data, Revision: rev})
|
|
}
|
|
}
|
|
return rev, nil
|
|
}
|
|
|
|
// clock is when a write lands: the store's own time, which a test sets.
|
|
func (s *fakeStore) clock() time.Time {
|
|
if s.now != nil {
|
|
return s.now()
|
|
}
|
|
return t0
|
|
}
|
|
|
|
func (s *fakeStore) Put(key string, value any) (uint64, error) { return s.write(key, value, "put") }
|
|
|
|
func (s *fakeStore) Create(key string, value any) (uint64, error) {
|
|
if e, _ := s.Get(key); e != nil {
|
|
return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
|
|
}
|
|
return s.write(key, value, "create")
|
|
}
|
|
|
|
func (s *fakeStore) Update(key string, value any, revision uint64) (uint64, error) {
|
|
if e, _ := s.Get(key); e == nil || e.Revision != revision {
|
|
return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
|
|
}
|
|
return s.write(key, value, "update")
|
|
}
|
|
|
|
func (s *fakeStore) Delete(key string) error {
|
|
s.mu.Lock()
|
|
delete(s.values, key)
|
|
delete(s.revs, key)
|
|
s.rev++
|
|
s.writes = append(s.writes, "delete "+key)
|
|
rev, ws := s.rev, append([]fakeWatch(nil), s.watchers...)
|
|
s.mu.Unlock()
|
|
for _, w := range ws {
|
|
if matchKey(w.key, key) {
|
|
w.handle(stdio.StateChange{State: s.name, Key: key, Op: "delete", Revision: rev})
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *fakeStore) Keys() ([]string, error) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
keys := []string{}
|
|
for k := range s.values {
|
|
keys = append(keys, k)
|
|
}
|
|
sortStrings(keys)
|
|
return keys, nil
|
|
}
|
|
|
|
func (s *fakeStore) Watch(key string, handle func(stdio.StateChange) error) error {
|
|
keys, _ := s.Keys()
|
|
for _, k := range keys {
|
|
if !matchKey(key, k) {
|
|
continue
|
|
}
|
|
e, _ := s.Get(k)
|
|
if err := handle(stdio.StateChange{State: s.name, Key: k, Op: "put", Value: e.Value, Revision: e.Revision, Current: true}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
s.mu.Lock()
|
|
s.watchers = append(s.watchers, fakeWatch{key, handle})
|
|
s.mu.Unlock()
|
|
return nil
|
|
}
|
|
|
|
// Writes is what was written, in order, and forgets it.
|
|
func (s *fakeStore) Writes() []string {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
w := s.writes
|
|
s.writes = nil
|
|
return w
|
|
}
|
|
|
|
// matchKey narrows as the bus does: `*` one name, `**` the rest, empty every key.
|
|
func matchKey(pattern, key string) bool {
|
|
if pattern == "" {
|
|
return true
|
|
}
|
|
p, k := strings.Split(pattern, "."), strings.Split(key, ".")
|
|
for i := range p {
|
|
if p[i] == "**" {
|
|
return len(k) > i
|
|
}
|
|
if i >= len(k) || (p[i] != "*" && p[i] != k[i]) {
|
|
return false
|
|
}
|
|
}
|
|
return len(p) == len(k)
|
|
}
|
|
|
|
func sortStrings(s []string) {
|
|
for i := 1; i < len(s); i++ {
|
|
for j := i; j > 0 && s[j] < s[j-1]; j-- {
|
|
s[j], s[j-1] = s[j-1], s[j]
|
|
}
|
|
}
|
|
}
|
|
|
|
// fakeStates is the module's states in memory.
|
|
type fakeStates struct {
|
|
mu sync.Mutex
|
|
states map[string]*fakeStore
|
|
}
|
|
|
|
func (f *fakeStates) State(name string) Store { return f.store(name) }
|
|
|
|
func (f *fakeStates) store(name string) *fakeStore {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
if f.states == nil {
|
|
f.states = map[string]*fakeStore{}
|
|
}
|
|
if f.states[name] == nil {
|
|
f.states[name] = newFakeStore(name)
|
|
}
|
|
return f.states[name]
|
|
}
|
|
|
|
// emitted is the events a keeper emitted.
|
|
type emitted struct {
|
|
mu sync.Mutex
|
|
events []emittedEvent
|
|
}
|
|
|
|
type emittedEvent struct {
|
|
name string
|
|
body any
|
|
id string
|
|
}
|
|
|
|
func (e *emitted) Emit(name string, body any, opts ...stdio.EmitOptions) error {
|
|
e.mu.Lock()
|
|
defer e.mu.Unlock()
|
|
id := ""
|
|
for _, o := range opts {
|
|
id = o.MessageID
|
|
}
|
|
e.events = append(e.events, emittedEvent{name, body, id})
|
|
return nil
|
|
}
|
|
|
|
func (e *emitted) Take() []emittedEvent {
|
|
e.mu.Lock()
|
|
defer e.mu.Unlock()
|
|
out := e.events
|
|
e.events = nil
|
|
return out
|
|
}
|
|
|
|
// fakeClock moves only when the test moves it; After fires once the clock passes it.
|
|
type fakeClock struct {
|
|
mu sync.Mutex
|
|
now time.Time
|
|
timers []fakeTimer
|
|
}
|
|
|
|
type fakeTimer struct {
|
|
at time.Time
|
|
c chan time.Time
|
|
}
|
|
|
|
func (c *fakeClock) Now() time.Time {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
return c.now
|
|
}
|
|
|
|
func (c *fakeClock) After(d time.Duration) <-chan time.Time {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
t := fakeTimer{c.now.Add(d), make(chan time.Time, 1)}
|
|
if d <= 0 {
|
|
t.c <- c.now
|
|
return t.c
|
|
}
|
|
c.timers = append(c.timers, t)
|
|
return t.c
|
|
}
|
|
|
|
func (c *fakeClock) Advance(d time.Duration) {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
c.now = c.now.Add(d)
|
|
kept := c.timers[:0]
|
|
for _, t := range c.timers {
|
|
if !t.at.After(c.now) {
|
|
t.c <- c.now
|
|
continue
|
|
}
|
|
kept = append(kept, t)
|
|
}
|
|
c.timers = kept
|
|
}
|
|
|
|
// fakeWatcher hands the test's signals to the keeper.
|
|
type fakeWatcher struct{ c chan Signal }
|
|
|
|
func (w fakeWatcher) Signals() <-chan Signal { return w.c }
|
|
func (w fakeWatcher) Close() error { return nil }
|
|
|
|
// pacmanDB is a node's package database as pacman answers it: the test changes it between reads.
|
|
type pacmanDB struct {
|
|
mu sync.Mutex
|
|
q, qe string
|
|
sl string
|
|
checkup Ran
|
|
commands []string
|
|
}
|
|
|
|
func (d *pacmanDB) set(q, qe string) {
|
|
d.mu.Lock()
|
|
defer d.mu.Unlock()
|
|
d.q, d.qe = q, qe
|
|
}
|
|
|
|
func (d *pacmanDB) setUpdates(r Ran) {
|
|
d.mu.Lock()
|
|
defer d.mu.Unlock()
|
|
d.checkup = r
|
|
}
|
|
|
|
func (d *pacmanDB) runner() Runner {
|
|
return func(_ context.Context, name string, args ...string) Ran {
|
|
d.mu.Lock()
|
|
defer d.mu.Unlock()
|
|
line := strings.TrimSpace(name + " " + strings.Join(args, " "))
|
|
d.commands = append(d.commands, line)
|
|
switch line {
|
|
case "pacman -Q":
|
|
return Ran{Stdout: d.q}
|
|
case "pacman -Qeq":
|
|
return Ran{Stdout: d.qe}
|
|
case "pacman -Sl":
|
|
return Ran{Stdout: d.sl}
|
|
case "checkupdates":
|
|
return d.checkup
|
|
}
|
|
return Ran{Status: 99, Stderr: "unexpected command: " + line}
|
|
}
|
|
}
|
|
|
|
var syncDBs = "core glibc 2.42-1 [installed]\ncore linux 6.17.1-1 [installed]\nextra zsh 5.9-2 [installed: 5.9-1]\n" +
|
|
"extra zsh-completions 0.35-1\nextra tmux 3.5-1\nextra-testing zsh 5.9-3\n"
|
|
|
|
var t0 = time.Date(2026, 10, 24, 4, 0, 0, 0, time.UTC)
|
|
|
|
func newTestKeeper(db *pacmanDB) (*Keeper, *fakeStates, *emitted, *fakeClock, fakeWatcher) {
|
|
states, ev, clock := &fakeStates{}, &emitted{}, &fakeClock{now: t0}
|
|
w := fakeWatcher{make(chan Signal)}
|
|
m := machine(db.runner(), 1000)
|
|
k := &Keeper{Node: "node-a", Machine: m, State: states.State, Events: ev, Clock: clock, Watcher: w,
|
|
Logf: func(string, ...any) {}}
|
|
return k, states, ev, clock, w
|
|
}
|
|
|
|
func installedOf(t *testing.T, s *fakeStore, key string) InstalledValue {
|
|
t.Helper()
|
|
e, _ := s.Get(key)
|
|
if e == nil {
|
|
t.Fatalf("no %s", key)
|
|
}
|
|
var v InstalledValue
|
|
if err := json.Unmarshal(e.Value, &v); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return v
|
|
}
|
|
|
|
// 1840/AC1: at the start every package is written with its version, its reason and its repository or foreign,
|
|
// and written again only when it differs from the key.
|
|
func TestAtItsStartTheKeeperWritesEveryPackageOnlyWhenItDiffersFromTheKey(t *testing.T) {
|
|
db := &pacmanDB{sl: syncDBs, checkup: Ran{Status: 2}}
|
|
db.set("glibc 2.42-1\nlinux 6.17.1-1\nyay 12.0-1\nzsh 5.9-1\n", "linux\nyay\nzsh\n")
|
|
k, states, ev, _, _ := newTestKeeper(db)
|
|
k.DBChanged = func() time.Time { return t0.Add(-time.Hour) }
|
|
k.refreshInstalled(k.DBChanged())
|
|
inst := states.store(StateInstalled)
|
|
if w := inst.Writes(); strings.Join(w, ",") != "put node-a" {
|
|
t.Fatalf("writes: %v", w)
|
|
}
|
|
v := installedOf(t, inst, "node-a")
|
|
want := []InstalledPackage{
|
|
{"glibc", "2.42-1", false, "core"}, {"linux", "6.17.1-1", true, "core"},
|
|
{"yay", "12.0-1", true, RepoForeign}, {"zsh", "5.9-1", true, "extra"},
|
|
}
|
|
if !samePackages(v.Packages, want) || !v.At.Equal(t0.Add(-time.Hour)) {
|
|
t.Fatalf("installed.node-a: %+v", v)
|
|
}
|
|
if e := ev.Take(); len(e) != 1 || len(e[0].body.(PackageChanged).Added) != 4 {
|
|
t.Fatalf("the first write adds every package: %+v", e)
|
|
}
|
|
// a second start over the same database writes nothing and says nothing, whatever the time.
|
|
k.refreshInstalled(t0)
|
|
if w := inst.Writes(); len(w) != 0 {
|
|
t.Errorf("the same packages were written again: %v", w)
|
|
}
|
|
if e := ev.Take(); len(e) != 0 {
|
|
t.Errorf("the same packages were said again: %+v", e)
|
|
}
|
|
}
|
|
|
|
// 1840/AC2: a transaction that adds, removes and changes packages is one package.changed naming the node and
|
|
// exactly those; one that changes nothing writes and emits nothing; a lock held again waits for its release.
|
|
func TestAReleasedLockSaysExactlyWhatTheTransactionChangedOnce(t *testing.T) {
|
|
db := &pacmanDB{sl: syncDBs, checkup: Ran{Status: 2}}
|
|
db.set("glibc 2.42-1\nlinux 6.17.1-1\nzsh 5.9-1\n", "linux\nzsh\n")
|
|
k, states, ev, clock, _ := newTestKeeper(db)
|
|
k.refreshInstalled(t0)
|
|
ev.Take()
|
|
states.store(StateInstalled).Writes()
|
|
|
|
held := false
|
|
k.Locked = func() bool { return held }
|
|
db.set("glibc 2.42-1\nlinux 6.17.2-1\ntmux 3.5-1\n", "linux\ntmux\n")
|
|
held = true
|
|
k.signal(LockReleased)
|
|
if w := states.store(StateInstalled).Writes(); len(w) != 0 {
|
|
t.Fatalf("read while the lock was held again: %v", w)
|
|
}
|
|
held = false
|
|
clock.Advance(time.Minute)
|
|
k.signal(LockReleased)
|
|
if w := states.store(StateInstalled).Writes(); strings.Join(w, ",") != "put node-a" {
|
|
t.Fatalf("writes: %v", w)
|
|
}
|
|
e := ev.Take()
|
|
if len(e) != 1 || e[0].name != EventPackageChanged {
|
|
t.Fatalf("events: %+v", e)
|
|
}
|
|
got := e[0].body.(PackageChanged)
|
|
if got.Node != "node-a" || !got.At.Equal(t0.Add(time.Minute)) ||
|
|
fmt.Sprint(got.Added) != "[{tmux 3.5-1}]" || fmt.Sprint(got.Removed) != "[{zsh 5.9-1}]" ||
|
|
fmt.Sprint(got.Changed) != "[{linux 6.17.1-1 6.17.2-1}]" {
|
|
t.Fatalf("package.changed: %+v", got)
|
|
}
|
|
if !strings.HasPrefix(e[0].id, "installed.node-a.") {
|
|
t.Errorf("the event's id is the key's revision, so a second emit is dropped: %q", e[0].id)
|
|
}
|
|
// a transaction that changed nothing (a database sync alone).
|
|
k.signal(LockReleased)
|
|
if w := states.store(StateInstalled).Writes(); len(w) != 0 {
|
|
t.Errorf("nothing changed and the key was written: %v", w)
|
|
}
|
|
if e := ev.Take(); len(e) != 0 {
|
|
t.Errorf("nothing changed and it was said: %+v", e)
|
|
}
|
|
// a change of reason alone rewrites the key and is no package.changed.
|
|
db.set("glibc 2.42-1\nlinux 6.17.2-1\ntmux 3.5-1\n", "glibc\nlinux\ntmux\n")
|
|
k.signal(LockReleased)
|
|
if w := states.store(StateInstalled).Writes(); len(w) != 1 {
|
|
t.Errorf("a changed reason is a changed key: %v", w)
|
|
}
|
|
if e := ev.Take(); len(e) != 0 {
|
|
t.Errorf("a changed reason is no change of a version: %+v", e)
|
|
}
|
|
}
|
|
|
|
// bigDB is a package database whose value is over the bus's cap.
|
|
func bigDB(n int) string {
|
|
var b strings.Builder
|
|
for i := 0; i < n; i++ {
|
|
fmt.Fprintf(&b, "package-with-a-long-name-%05d 1.0.0-1\n", i)
|
|
}
|
|
return b.String()
|
|
}
|
|
|
|
// 1840/AC3: a value over the bus's cap is not written: the key keeps its last value and `<node>.not-current`
|
|
// holds the size and since when, written once; the next write that fits deletes it.
|
|
func TestAValueOverTheCapKeepsTheLastAndMarksTheNodeNotCurrent(t *testing.T) {
|
|
db := &pacmanDB{sl: syncDBs, checkup: Ran{Status: 2}}
|
|
db.set("glibc 2.42-1\n", "")
|
|
k, states, ev, clock, _ := newTestKeeper(db)
|
|
k.refreshInstalled(t0)
|
|
inst := states.store(StateInstalled)
|
|
inst.Writes()
|
|
ev.Take()
|
|
|
|
db.set(bigDB(5000), "")
|
|
clock.Advance(time.Minute)
|
|
k.signal(LockReleased)
|
|
if w := inst.Writes(); strings.Join(w, ",") != "put node-a.not-current" {
|
|
t.Fatalf("writes: %v", w)
|
|
}
|
|
if v := installedOf(t, inst, "node-a"); len(v.Packages) != 1 {
|
|
t.Fatalf("the last value is kept: %d packages", len(v.Packages))
|
|
}
|
|
if e := ev.Take(); len(e) != 0 {
|
|
t.Errorf("nothing was written and it was said: %+v", e)
|
|
}
|
|
e, _ := inst.Get("node-a.not-current")
|
|
var nc NotCurrent
|
|
json.Unmarshal(e.Value, &nc)
|
|
if nc.Bytes <= ValueCap || !nc.Since.Equal(t0.Add(time.Minute)) {
|
|
t.Fatalf("not-current: %+v", nc)
|
|
}
|
|
// still over: the mark is written once.
|
|
db.set(bigDB(5001), "")
|
|
clock.Advance(time.Minute)
|
|
k.signal(LockReleased)
|
|
if w := inst.Writes(); len(w) != 0 {
|
|
t.Errorf("the mark was written again: %v", w)
|
|
}
|
|
// back under the cap: written, and the mark deleted.
|
|
db.set("glibc 2.42-1\nzsh 5.9-1\n", "")
|
|
k.signal(LockReleased)
|
|
if w := inst.Writes(); strings.Join(w, ",") != "put node-a,delete node-a.not-current" {
|
|
t.Errorf("writes: %v", w)
|
|
}
|
|
}
|
|
|
|
// The mark goes too when the node's packages come back to what the key holds: the key is current again.
|
|
func TestTheMarkGoesWhenThePackagesComeBackToTheKey(t *testing.T) {
|
|
db := &pacmanDB{sl: syncDBs, checkup: Ran{Status: 2}}
|
|
db.set("glibc 2.42-1\n", "")
|
|
k, states, _, _, _ := newTestKeeper(db)
|
|
k.refreshInstalled(t0)
|
|
db.set(bigDB(5000), "")
|
|
k.signal(LockReleased)
|
|
inst := states.store(StateInstalled)
|
|
inst.Writes()
|
|
db.set("glibc 2.42-1\n", "")
|
|
k.signal(LockReleased)
|
|
if w := inst.Writes(); strings.Join(w, ",") != "delete node-a.not-current" {
|
|
t.Errorf("writes: %v", w)
|
|
}
|
|
}
|
|
|
|
// The loop: a released lock is read, a hook runs after it with what it changed, a watched state's change reaches
|
|
// its hook, and a wake comes at the time it asked.
|
|
func TestTheLoopRunsTheHooksOnItsOneGoroutine(t *testing.T) {
|
|
db := &pacmanDB{sl: syncDBs, checkup: Ran{Status: 2}}
|
|
db.set("glibc 2.42-1\n", "")
|
|
k, states, _, clock, w := newTestKeeper(db)
|
|
var mu sync.Mutex
|
|
var seen []string
|
|
say := func(s string) { mu.Lock(); seen = append(seen, s); mu.Unlock() }
|
|
k.Hooks = Hooks{
|
|
Start: []func(*Keeper) error{func(*Keeper) error { say("start"); return nil }},
|
|
AfterTransaction: []func(*Keeper, Transaction) error{func(_ *Keeper, tr Transaction) error {
|
|
say(fmt.Sprintf("transaction changed=%v", tr.Changed != nil))
|
|
return nil
|
|
}},
|
|
Changes: []StateHook{{State: StateAsks, Key: "node-a.*", Handle: func(_ *Keeper, c stdio.StateChange) error {
|
|
say(fmt.Sprintf("%s %s current=%v", c.Op, c.Key, c.Current))
|
|
return nil
|
|
}}},
|
|
Wakes: []Wake{func(_ *Keeper, now time.Time) time.Time {
|
|
if now.Equal(t0) {
|
|
return t0.Add(time.Hour)
|
|
}
|
|
say("wake " + now.Sub(t0).String())
|
|
return time.Time{}
|
|
}},
|
|
}
|
|
states.store(StateAsks).Put("node-a.ask-1", Ask{ID: "ask-1"})
|
|
states.store(StateAsks).Put("node-b.ask-2", Ask{ID: "ask-2"})
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan error, 1)
|
|
go func() { done <- k.Run(ctx) }()
|
|
waitFor(t, &mu, &seen, "start")
|
|
db.set("glibc 2.42-1\nzsh 5.9-1\n", "")
|
|
w.c <- LockReleased
|
|
waitFor(t, &mu, &seen, "transaction changed=true")
|
|
states.store(StateAsks).Put("node-a.ask-3", Ask{ID: "ask-3"})
|
|
waitFor(t, &mu, &seen, "put node-a.ask-3 current=false")
|
|
clock.Advance(time.Hour)
|
|
waitFor(t, &mu, &seen, "wake 1h0m0s")
|
|
cancel()
|
|
if err := <-done; err != context.Canceled {
|
|
t.Errorf("Run ended with %v", err)
|
|
}
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
if strings.Join(seen, "|") != "put node-a.ask-1 current=true|start|transaction changed=true|put node-a.ask-3 current=false|wake 1h0m0s" &&
|
|
strings.Join(seen, "|") != "start|put node-a.ask-1 current=true|transaction changed=true|put node-a.ask-3 current=false|wake 1h0m0s" {
|
|
t.Errorf("seen: %v", seen)
|
|
}
|
|
}
|
|
|
|
func waitFor(t *testing.T, mu *sync.Mutex, seen *[]string, want string) {
|
|
t.Helper()
|
|
for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(time.Millisecond) {
|
|
mu.Lock()
|
|
for _, s := range *seen {
|
|
if s == want {
|
|
mu.Unlock()
|
|
return
|
|
}
|
|
}
|
|
mu.Unlock()
|
|
}
|
|
t.Fatalf("never saw %q", want)
|
|
}
|
|
|
|
// The watch ending ends the keeper, loudly.
|
|
func TestTheKeeperStopsWhenTheWatchEnds(t *testing.T) {
|
|
db := &pacmanDB{sl: syncDBs, checkup: Ran{Status: 2}}
|
|
k, _, _, _, w := newTestKeeper(db)
|
|
close(w.c)
|
|
if err := k.Run(context.Background()); err == nil || !strings.Contains(err.Error(), "watch ended") {
|
|
t.Errorf("Run: %v", err)
|
|
}
|
|
}
|
|
|
|
// The machine's watch: the lock deleted and the mirror list written or moved in are each a signal; anything else
|
|
// in those directories is none.
|
|
func TestTheWatchSeesTheLockReleasedAndTheMirrorListRewritten(t *testing.T) {
|
|
lockDir, mirrorDir := t.TempDir(), t.TempDir()
|
|
w, err := WatchMachine(lockDir, LockName, mirrorDir, MirrorName)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
next := func() Signal {
|
|
select {
|
|
case s := <-w.Signals():
|
|
return s
|
|
case <-time.After(200 * time.Millisecond):
|
|
return 0
|
|
}
|
|
}
|
|
lock := filepath.Join(lockDir, LockName)
|
|
os.WriteFile(lock, nil, 0o644)
|
|
os.WriteFile(filepath.Join(lockDir, "other"), nil, 0o644)
|
|
os.Remove(filepath.Join(lockDir, "other"))
|
|
if s := next(); s != 0 {
|
|
t.Fatalf("a lock taken, and another file, are no signal: %v", s)
|
|
}
|
|
os.Remove(lock)
|
|
if s := next(); s != LockReleased {
|
|
t.Fatalf("the lock deleted: %v", s)
|
|
}
|
|
os.WriteFile(filepath.Join(mirrorDir, MirrorName), []byte("Server = x\n"), 0o644)
|
|
if s := next(); s != MirrorsRewritten {
|
|
t.Fatalf("the list written: %v", s)
|
|
}
|
|
tmp := filepath.Join(mirrorDir, "mirrorlist.new")
|
|
os.WriteFile(tmp, []byte("Server = y\n"), 0o644)
|
|
if s := next(); s != 0 {
|
|
t.Fatalf("another file written: %v", s)
|
|
}
|
|
os.Rename(tmp, filepath.Join(mirrorDir, MirrorName))
|
|
if s := next(); s != MirrorsRewritten {
|
|
t.Fatalf("the list moved in: %v", s)
|
|
}
|
|
w.Close()
|
|
for deadline := time.After(2 * time.Second); ; {
|
|
select {
|
|
case _, open := <-w.Signals():
|
|
if !open {
|
|
return
|
|
}
|
|
case <-deadline:
|
|
t.Fatal("closing the watch did not end its signals")
|
|
}
|
|
}
|
|
}
|