A machine whose presence says online is heard now; last heard is otherwise the later of last_seen and the presence time (issue 1156)

This commit is contained in:
2026-10-18 21:04:59 +02:00
parent 7be2df627f
commit fb66608a13
4 changed files with 61 additions and 27 deletions
+33 -13
View File
@@ -35,10 +35,17 @@ const (
presenceSilent presenceSilent
) )
// applyPresence sets what the watchman says of a machine: the verdict of its stuck presence key (nil when there is none) // presenceEntry is presence.<node> as the watchman wrote it: whether the machine is online and when the entry was made.
// and, when presence.<node> has an entry, its time as when the machine was last heard from. Without a presence entry, // The watchman writes it only when it changes, so for an online machine the time is when it connected.
// last heard stays the store's last_seen. type presenceEntry struct {
func applyPresence(m *machineFacts, stuck *sdkdelivery.Stuck, at time.Time) { Online bool
At time.Time
}
// applyPresence sets what the watchman says of a machine: the verdict of its stuck presence key (nil when there is none).
// A presence entry that says online counts the machine as heard now; otherwise last heard is the later of the store's
// last_seen and the entry's time (the time it went offline).
func applyPresence(m *machineFacts, stuck *sdkdelivery.Stuck, entry *presenceEntry, now time.Time) {
switch { switch {
case stuck == nil: case stuck == nil:
m.presence = presenceUnsaid m.presence = presenceUnsaid
@@ -47,8 +54,12 @@ func applyPresence(m *machineFacts, stuck *sdkdelivery.Stuck, at time.Time) {
default: default:
m.presence = presenceSilent m.presence = presenceSilent
} }
if !at.IsZero() { switch {
m.lastHeard = at case entry == nil:
case entry.Online:
m.lastHeard = now
case entry.At.After(m.lastHeard):
m.lastHeard = entry.At
} }
} }
@@ -137,8 +148,8 @@ func readStuck(ctx context.Context, conn *nats.Conn) (map[string]sdkdelivery.Stu
return out, nil return out, nil
} }
// readPresenceTimes is, by node, the time of presence.<node>'s entry; a node without an entry is left out. // readPresence is, by node, presence.<node>'s entry; a node without an entry is left out.
func readPresenceTimes(ctx context.Context, conn *nats.Conn, nodes []string) (map[string]time.Time, error) { func readPresence(ctx context.Context, conn *nats.Conn, nodes []string) (map[string]presenceEntry, error) {
api, err := jetstream.New(conn) api, err := jetstream.New(conn)
if err != nil { if err != nil {
return nil, err return nil, err
@@ -150,7 +161,7 @@ func readPresenceTimes(ctx context.Context, conn *nats.Conn, nodes []string) (ma
if err != nil { if err != nil {
return nil, fmt.Errorf("the watchman's presence bucket: %w", err) return nil, fmt.Errorf("the watchman's presence bucket: %w", err)
} }
out := map[string]time.Time{} out := map[string]presenceEntry{}
for _, node := range nodes { for _, node := range nodes {
key, err := sdkdelivery.PresenceNodeKey(node) key, err := sdkdelivery.PresenceNodeKey(node)
if err != nil { if err != nil {
@@ -163,13 +174,18 @@ func readPresenceTimes(ctx context.Context, conn *nats.Conn, nodes []string) (ma
if err != nil { if err != nil {
return nil, err return nil, err
} }
out[node] = e.Created() p, err := sdkdelivery.DecodePresence(e.Value())
if err != nil {
fmt.Printf("the presence key %s is not read: %v\n", key, err)
continue
}
out[node] = presenceEntry{Online: p.Online, At: e.Created()}
} }
return out, nil return out, nil
} }
// withPresence says of each machine what the watchman says: its stuck presence key and its presence entry's time. // withPresence says of each machine what the watchman says: its stuck presence key and its presence entry's time.
func withPresence(ctx context.Context, conn *nats.Conn, machines []machineFacts) error { func withPresence(ctx context.Context, conn *nats.Conn, machines []machineFacts, now time.Time) error {
stuck, err := readStuck(ctx, conn) stuck, err := readStuck(ctx, conn)
if err != nil { if err != nil {
return err return err
@@ -178,7 +194,7 @@ func withPresence(ctx context.Context, conn *nats.Conn, machines []machineFacts)
for i, m := range machines { for i, m := range machines {
names[i] = m.name names[i] = m.name
} }
times, err := readPresenceTimes(ctx, conn, names) entries, err := readPresence(ctx, conn, names)
if err != nil { if err != nil {
return err return err
} }
@@ -189,7 +205,11 @@ func withPresence(ctx context.Context, conn *nats.Conn, machines []machineFacts)
said = &s said = &s
} }
} }
applyPresence(&machines[i], said, times[machines[i].name]) var entry *presenceEntry
if e, ok := entries[machines[i].name]; ok {
entry = &e
}
applyPresence(&machines[i], said, entry, now)
} }
return nil return nil
} }
@@ -61,7 +61,7 @@ func TestAStuckKeyWrittenAndThenMovedRaisesAndClearsACondition(t *testing.T) { /
t.Fatalf("a stuck key written: %+v %v", got, err) t.Fatalf("a stuck key written: %+v %v", got, err)
} }
machines := []machineFacts{{name: "ace", lastHeard: time.Now()}} machines := []machineFacts{{name: "ace", lastHeard: time.Now()}}
if err := withPresence(ctx, js.Conn(), machines); err != nil { if err := withPresence(ctx, js.Conn(), machines, time.Now()); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if !machines[0].silent(time.Now()) { if !machines[0].silent(time.Now()) {
@@ -73,7 +73,7 @@ func TestAStuckKeyWrittenAndThenMovedRaisesAndClearsACondition(t *testing.T) { /
t.Fatalf("a stuck key written as moved: %+v %v", got, err) t.Fatalf("a stuck key written as moved: %+v %v", got, err)
} }
machines = []machineFacts{{name: "ace", lastHeard: time.Now().Add(-time.Hour), every: time.Minute}} machines = []machineFacts{{name: "ace", lastHeard: time.Now().Add(-time.Hour), every: time.Minute}}
if err := withPresence(ctx, js.Conn(), machines); err != nil { if err := withPresence(ctx, js.Conn(), machines, time.Now()); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if machines[0].silent(time.Now()) { if machines[0].silent(time.Now()) {
+25 -11
View File
@@ -70,7 +70,7 @@ func TestAStuckPresenceKeyIsTheSilentVerdict(t *testing.T) { // 1156/AC2
f := calm(now) f := calm(now)
f.machines = append(f.machines, aMachineHeardAt(now, 10*time.Second)) f.machines = append(f.machines, aMachineHeardAt(now, 10*time.Second))
notMoved := sdkdelivery.Stuck{Bucket: sdkdelivery.PresenceBucket, Key: "laptop", Waited: "its return", Since: now.Add(-3 * time.Minute)} notMoved := sdkdelivery.Stuck{Bucket: sdkdelivery.PresenceBucket, Key: "laptop", Waited: "its return", Since: now.Add(-3 * time.Minute)}
applyPresence(&f.machines[1], &notMoved, now.Add(-3*time.Minute)) applyPresence(&f.machines[1], &notMoved, nil, now)
got := watchHeartbeats(f) got := watchHeartbeats(f)
if len(got) != 1 || got[0].Machine != "laptop" || got[0].Kind != "silent" { if len(got) != 1 || got[0].Machine != "laptop" || got[0].Kind != "silent" {
t.Fatalf("a presence key not moved: %+v", got) t.Fatalf("a presence key not moved: %+v", got)
@@ -85,24 +85,38 @@ func TestAStuckPresenceKeyIsTheSilentVerdict(t *testing.T) { // 1156/AC2
f.machines[1].lastHeard = now.Add(-2 * time.Hour) f.machines[1].lastHeard = now.Add(-2 * time.Hour)
moved := notMoved moved := notMoved
moved.Moved = true moved.Moved = true
applyPresence(&f.machines[1], &moved, now.Add(-time.Minute)) applyPresence(&f.machines[1], &moved, nil, now)
if got := watchHeartbeats(f); len(got) != 0 { if got := watchHeartbeats(f); len(got) != 0 {
t.Fatalf("a presence key that moved raised %+v", got) t.Fatalf("a presence key that moved raised %+v", got)
} }
} }
// 1156/AC3: a machine's last heard is the presence key's entry time when there is one, and last_seen when there is none. // 1156/AC3: with a presence entry that says offline, last heard is the later of last_seen and the entry's time; with none,
func TestLastHeardIsThePresenceKeysTimeWhenThereIsOne(t *testing.T) { // 1156/AC3 // last_seen. An entry that says online counts as heard now: the watchman writes it only when it changes, so its time is
// when the machine connected, and a machine online for hours with no stuck key is not silent, whatever last_seen says.
func TestLastHeardFollowsThePresenceEntry(t *testing.T) { // 1156/AC3
now := time.Now() now := time.Now()
seen := now.Add(-time.Hour) seen := now.Add(-time.Hour)
m := machineFacts{name: "laptop", lastHeard: seen} m := machineFacts{name: "laptop", lastHeard: seen, every: time.Minute}
applyPresence(&m, nil, time.Time{}) applyPresence(&m, nil, nil, now)
if !m.lastHeard.Equal(seen) { if !m.lastHeard.Equal(seen) {
t.Fatalf("with no presence key last heard is %v, want last_seen %v", m.lastHeard, seen) t.Fatalf("with no presence entry last heard is %v, want last_seen %v", m.lastHeard, seen)
} }
at := now.Add(-5 * time.Minute) offlineAt := now.Add(-5 * time.Minute)
applyPresence(&m, nil, at) applyPresence(&m, nil, &presenceEntry{Online: false, At: offlineAt}, now)
if !m.lastHeard.Equal(at) { if !m.lastHeard.Equal(offlineAt) {
t.Fatalf("with a presence key last heard is %v, want %v", m.lastHeard, at) t.Fatalf("offline since %v: last heard is %v", offlineAt, m.lastHeard)
}
m = machineFacts{name: "laptop", lastHeard: now.Add(-time.Minute), every: time.Minute}
applyPresence(&m, nil, &presenceEntry{Online: false, At: now.Add(-time.Hour)}, now)
if !m.lastHeard.Equal(now.Add(-time.Minute)) {
t.Fatalf("an older entry replaced a newer last_seen: %v", m.lastHeard)
}
for _, lastSeen := range []time.Duration{10 * time.Second, 5 * time.Hour} {
m := machineFacts{name: "laptop", lastHeard: now.Add(-lastSeen), every: time.Minute}
applyPresence(&m, nil, &presenceEntry{Online: true, At: now.Add(-6 * time.Hour)}, now)
if m.silent(now) {
t.Fatalf("a machine online since hours ago (last_seen %v ago) with no stuck key is silent", lastSeen)
}
} }
} }
+1 -1
View File
@@ -483,7 +483,7 @@ func (w *watchdogs) gatherMachines(ctx context.Context, inv *inventory.Inventory
// What the watchman says of each machine: its stuck presence key is the silent verdict, and presence.<node>'s time when // What the watchman says of each machine: its stuck presence key is the silent verdict, and presence.<node>'s time when
// it was last heard (ADR 0433 §3). Unreadable is said, and the heartbeat rule judges as before. // it was last heard (ADR 0433 §3). Unreadable is said, and the heartbeat rule judges as before.
if w.js != nil { if w.js != nil {
if err := withPresence(ctx, w.js.Conn(), out); err != nil { if err := withPresence(ctx, w.js.Conn(), out, now); err != nil {
fmt.Printf("what the watchman says of the machines' presence cannot be read: %v\n", err) fmt.Printf("what the watchman says of the machines' presence cannot be read: %v\n", err)
} else { } else {
for _, m := range out { for _, m := range out {