fix(brokerstore): zero tab attention before persisting snapshots (C3)
Store() already sanitized Checking/LastFailure.Err/Policy before writing the durable snapshot, but left tab Attention untouched. A bell recorded the instant before a broker restart would otherwise resurrect on reload even though nothing is still ringing. Zero Sessions[].Tabs[].Attention alongside the existing sanitization, and add a table test covering single, mixed, and no-attention tab lists across a store/reload cycle. Ports the intent of main's internal/adapters/remote/catalog_cache_test.go:71.
- commit
- 8e35ea9bae97a89f0f2f03a37c64ae1decbf2906
- author
- brice · · 2026-09-22 20:18 UTC
- parent
- 9a44abc
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +89 -0
M internal/adapters/brokerstore/store.go +9 -0
| @@ -322,6 +322,15 @@ func (s *Store) Store(snapshot ports.BrokerSnapshot) error { | |||
| 322 | 322 | snapshot.Daemons[i].Checking = false | |
| 323 | 323 | snapshot.Daemons[i].LastFailure.Err = nil | |
| 324 | 324 | snapshot.Daemons[i].Policy = ports.BrokerPolicy{} | |
| 325 | + | // Attention is transient observation state, not durable identity: a | |
| 326 | + | // bell recorded the instant before a broker restart must never | |
| 327 | + | // resurrect on reload, so every tab's Attention is zeroed before the | |
| 328 | + | // durable shape is validated and written. | |
| 329 | + | for j := range snapshot.Daemons[i].Sessions { | |
| 330 | + | for k := range snapshot.Daemons[i].Sessions[j].Tabs { | |
| 331 | + | snapshot.Daemons[i].Sessions[j].Tabs[k].Attention = false | |
| 332 | + | } | |
| 333 | + | } | |
| 325 | 334 | } | |
| 326 | 335 | if err := validateDurableSnapshot(snapshot); err != nil { | |
| 327 | 336 | return err | |
| @@ -322,6 +322,15 @@ func (s *Store) Store(snapshot ports.BrokerSnapshot) error { | |||||
| 322 | snapshot.Daemons[i].Checking = false | 322 | snapshot.Daemons[i].Checking = false | ||
| 323 | snapshot.Daemons[i].LastFailure.Err = nil | 323 | snapshot.Daemons[i].LastFailure.Err = nil | ||
| 324 | snapshot.Daemons[i].Policy = ports.BrokerPolicy{} | 324 | snapshot.Daemons[i].Policy = ports.BrokerPolicy{} | ||
| 325 | + | // Attention is transient observation state, not durable identity: a | |||
| 326 | + | // bell recorded the instant before a broker restart must never | |||
| 327 | + | // resurrect on reload, so every tab's Attention is zeroed before the | |||
| 328 | + | // durable shape is validated and written. | |||
| 329 | + | for j := range snapshot.Daemons[i].Sessions { | |||
| 330 | + | for k := range snapshot.Daemons[i].Sessions[j].Tabs { | |||
| 331 | + | snapshot.Daemons[i].Sessions[j].Tabs[k].Attention = false | |||
| 332 | + | } | |||
| 333 | + | } | |||
| 325 | } | 334 | } | ||
| 326 | if err := validateDurableSnapshot(snapshot); err != nil { | 335 | if err := validateDurableSnapshot(snapshot); err != nil { | ||
| 327 | return err | 336 | return err | ||
M internal/adapters/brokerstore/store_test.go +80 -0
| @@ -10,10 +10,12 @@ import ( | |||
| 10 | 10 | "sync" | |
| 11 | 11 | "syscall" | |
| 12 | 12 | "testing" | |
| 13 | + | "time" | |
| 13 | 14 | ||
| 14 | 15 | "github.com/bnema/vev/internal/domain" | |
| 15 | 16 | "github.com/bnema/vev/internal/ports" | |
| 16 | 17 | "github.com/bnema/vev/internal/protocol" | |
| 18 | + | "github.com/bnema/vev/internal/protocol/catalogue" | |
| 17 | 19 | "github.com/stretchr/testify/require" | |
| 18 | 20 | ) | |
| 19 | 21 | ||
| @@ -575,6 +577,84 @@ func TestDurableSnapshotDropsProcessLocalTombstones(t *testing.T) { | |||
| 575 | 577 | require.Empty(t, reopened.Removed) | |
| 576 | 578 | } | |
| 577 | 579 | ||
| 580 | + | // TestStoreZeroesTabAttentionOnPersist pins C3: a tab's Attention (bell) is | |
| 581 | + | // transient observation state, not durable identity, so it is zeroed before | |
| 582 | + | // every durable write. Without this, a bell recorded the instant before a | |
| 583 | + | // broker restart would resurrect on reload even though nothing is still | |
| 584 | + | // ringing. This ports the intent of main's | |
| 585 | + | // internal/adapters/remote/catalog_cache_test.go:71, which pins the same rule | |
| 586 | + | // for the legacy remote monitor's cache (there, dynamic tab fields including | |
| 587 | + | // Attention never reach the stored JSON at all). | |
| 588 | + | func TestStoreZeroesTabAttentionOnPersist(t *testing.T) { | |
| 589 | + | tests := []struct { | |
| 590 | + | name string | |
| 591 | + | tabs []catalogue.RemoteCatalogTab | |
| 592 | + | }{ | |
| 593 | + | {name: "single tab with attention", tabs: []catalogue.RemoteCatalogTab{ | |
| 594 | + | {ID: "t1", Index: 0, Attention: true}, | |
| 595 | + | }}, | |
| 596 | + | {name: "mixed attention across tabs", tabs: []catalogue.RemoteCatalogTab{ | |
| 597 | + | {ID: "t1", Index: 0, Attention: true}, | |
| 598 | + | {ID: "t2", Index: 1}, | |
| 599 | + | {ID: "t3", Index: 2, Attention: true}, | |
| 600 | + | }}, | |
| 601 | + | {name: "no attention stays false", tabs: []catalogue.RemoteCatalogTab{ | |
| 602 | + | {ID: "t1", Index: 0}, | |
| 603 | + | {ID: "t2", Index: 1}, | |
| 604 | + | }}, | |
| 605 | + | } | |
| 606 | + | for _, tt := range tests { | |
| 607 | + | t.Run(tt.name, func(t *testing.T) { | |
| 608 | + | o := options(t) | |
| 609 | + | s, err := Open(o) | |
| 610 | + | require.NoError(t, err) | |
| 611 | + | h, err := s.LoadHosts() | |
| 612 | + | require.NoError(t, err) | |
| 613 | + | require.Len(t, h.Hosts, 1) | |
| 614 | + | reg := h.Hosts[0].Registration | |
| 615 | + | ||
| 616 | + | snap, err := s.Load() | |
| 617 | + | require.NoError(t, err) | |
| 618 | + | snap.Epoch = 9 | |
| 619 | + | snap.Revision = 1 | |
| 620 | + | daemon := observation(reg.Endpoint, reg) | |
| 621 | + | daemon.InventoryKnown = true | |
| 622 | + | daemon.LastSuccess = time.Unix(1, 0) | |
| 623 | + | daemon.Sessions = []catalogue.RemoteCatalogSession{{ | |
| 624 | + | LifecycleID: domain.SessionLifecycleID{1}, | |
| 625 | + | Name: "work", | |
| 626 | + | State: catalogue.RemoteCatalogSessionUp, | |
| 627 | + | Tabs: tt.tabs, | |
| 628 | + | }} | |
| 629 | + | snap.Daemons = []ports.BrokerDaemonObservation{daemon} | |
| 630 | + | require.NoError(t, s.Store(snap)) | |
| 631 | + | ||
| 632 | + | stored, err := s.Load() | |
| 633 | + | require.NoError(t, err) | |
| 634 | + | require.Len(t, stored.Daemons, 1) | |
| 635 | + | require.Len(t, stored.Daemons[0].Sessions, 1) | |
| 636 | + | require.Len(t, stored.Daemons[0].Sessions[0].Tabs, len(tt.tabs)) | |
| 637 | + | for _, tab := range stored.Daemons[0].Sessions[0].Tabs { | |
| 638 | + | require.False(t, tab.Attention, "tab %q must never persist a live bell", tab.ID) | |
| 639 | + | } | |
| 640 | + | require.NoError(t, s.Close()) | |
| 641 | + | ||
| 642 | + | // The zeroing is durable, not just an in-memory view: a fresh Open | |
| 643 | + | // reading the same file back must never resurrect the bell either. | |
| 644 | + | s, err = Open(o) | |
| 645 | + | require.NoError(t, err) | |
| 646 | + | defer s.Close() | |
| 647 | + | reopened, err := s.Load() | |
| 648 | + | require.NoError(t, err) | |
| 649 | + | require.Len(t, reopened.Daemons, 1) | |
| 650 | + | require.Len(t, reopened.Daemons[0].Sessions, 1) | |
| 651 | + | for _, tab := range reopened.Daemons[0].Sessions[0].Tabs { | |
| 652 | + | require.False(t, tab.Attention) | |
| 653 | + | } | |
| 654 | + | }) | |
| 655 | + | } | |
| 656 | + | } | |
| 657 | + | ||
| 578 | 658 | func TestFencingAndCommitFaults(t *testing.T) { | |
| 579 | 659 | for _, point := range []string{"write", "sync", "rename", "dirsync"} { | |
| 580 | 660 | t.Run(point, func(t *testing.T) { | |
| @@ -10,10 +10,12 @@ import ( | |||||
| 10 | "sync" | 10 | "sync" | ||
| 11 | "syscall" | 11 | "syscall" | ||
| 12 | "testing" | 12 | "testing" | ||
| 13 | + | "time" | |||
| 13 | 14 | ||||
| 14 | "github.com/bnema/vev/internal/domain" | 15 | "github.com/bnema/vev/internal/domain" | ||
| 15 | "github.com/bnema/vev/internal/ports" | 16 | "github.com/bnema/vev/internal/ports" | ||
| 16 | "github.com/bnema/vev/internal/protocol" | 17 | "github.com/bnema/vev/internal/protocol" | ||
| 18 | + | "github.com/bnema/vev/internal/protocol/catalogue" | |||
| 17 | "github.com/stretchr/testify/require" | 19 | "github.com/stretchr/testify/require" | ||
| 18 | ) | 20 | ) | ||
| 19 | 21 | ||||
| @@ -575,6 +577,84 @@ func TestDurableSnapshotDropsProcessLocalTombstones(t *testing.T) { | |||||
| 575 | require.Empty(t, reopened.Removed) | 577 | require.Empty(t, reopened.Removed) | ||
| 576 | } | 578 | } | ||
| 577 | 579 | ||||
| 580 | + | // TestStoreZeroesTabAttentionOnPersist pins C3: a tab's Attention (bell) is | |||
| 581 | + | // transient observation state, not durable identity, so it is zeroed before | |||
| 582 | + | // every durable write. Without this, a bell recorded the instant before a | |||
| 583 | + | // broker restart would resurrect on reload even though nothing is still | |||
| 584 | + | // ringing. This ports the intent of main's | |||
| 585 | + | // internal/adapters/remote/catalog_cache_test.go:71, which pins the same rule | |||
| 586 | + | // for the legacy remote monitor's cache (there, dynamic tab fields including | |||
| 587 | + | // Attention never reach the stored JSON at all). | |||
| 588 | + | func TestStoreZeroesTabAttentionOnPersist(t *testing.T) { | |||
| 589 | + | tests := []struct { | |||
| 590 | + | name string | |||
| 591 | + | tabs []catalogue.RemoteCatalogTab | |||
| 592 | + | }{ | |||
| 593 | + | {name: "single tab with attention", tabs: []catalogue.RemoteCatalogTab{ | |||
| 594 | + | {ID: "t1", Index: 0, Attention: true}, | |||
| 595 | + | }}, | |||
| 596 | + | {name: "mixed attention across tabs", tabs: []catalogue.RemoteCatalogTab{ | |||
| 597 | + | {ID: "t1", Index: 0, Attention: true}, | |||
| 598 | + | {ID: "t2", Index: 1}, | |||
| 599 | + | {ID: "t3", Index: 2, Attention: true}, | |||
| 600 | + | }}, | |||
| 601 | + | {name: "no attention stays false", tabs: []catalogue.RemoteCatalogTab{ | |||
| 602 | + | {ID: "t1", Index: 0}, | |||
| 603 | + | {ID: "t2", Index: 1}, | |||
| 604 | + | }}, | |||
| 605 | + | } | |||
| 606 | + | for _, tt := range tests { | |||
| 607 | + | t.Run(tt.name, func(t *testing.T) { | |||
| 608 | + | o := options(t) | |||
| 609 | + | s, err := Open(o) | |||
| 610 | + | require.NoError(t, err) | |||
| 611 | + | h, err := s.LoadHosts() | |||
| 612 | + | require.NoError(t, err) | |||
| 613 | + | require.Len(t, h.Hosts, 1) | |||
| 614 | + | reg := h.Hosts[0].Registration | |||
| 615 | + | ||||
| 616 | + | snap, err := s.Load() | |||
| 617 | + | require.NoError(t, err) | |||
| 618 | + | snap.Epoch = 9 | |||
| 619 | + | snap.Revision = 1 | |||
| 620 | + | daemon := observation(reg.Endpoint, reg) | |||
| 621 | + | daemon.InventoryKnown = true | |||
| 622 | + | daemon.LastSuccess = time.Unix(1, 0) | |||
| 623 | + | daemon.Sessions = []catalogue.RemoteCatalogSession{{ | |||
| 624 | + | LifecycleID: domain.SessionLifecycleID{1}, | |||
| 625 | + | Name: "work", | |||
| 626 | + | State: catalogue.RemoteCatalogSessionUp, | |||
| 627 | + | Tabs: tt.tabs, | |||
| 628 | + | }} | |||
| 629 | + | snap.Daemons = []ports.BrokerDaemonObservation{daemon} | |||
| 630 | + | require.NoError(t, s.Store(snap)) | |||
| 631 | + | ||||
| 632 | + | stored, err := s.Load() | |||
| 633 | + | require.NoError(t, err) | |||
| 634 | + | require.Len(t, stored.Daemons, 1) | |||
| 635 | + | require.Len(t, stored.Daemons[0].Sessions, 1) | |||
| 636 | + | require.Len(t, stored.Daemons[0].Sessions[0].Tabs, len(tt.tabs)) | |||
| 637 | + | for _, tab := range stored.Daemons[0].Sessions[0].Tabs { | |||
| 638 | + | require.False(t, tab.Attention, "tab %q must never persist a live bell", tab.ID) | |||
| 639 | + | } | |||
| 640 | + | require.NoError(t, s.Close()) | |||
| 641 | + | ||||
| 642 | + | // The zeroing is durable, not just an in-memory view: a fresh Open | |||
| 643 | + | // reading the same file back must never resurrect the bell either. | |||
| 644 | + | s, err = Open(o) | |||
| 645 | + | require.NoError(t, err) | |||
| 646 | + | defer s.Close() | |||
| 647 | + | reopened, err := s.Load() | |||
| 648 | + | require.NoError(t, err) | |||
| 649 | + | require.Len(t, reopened.Daemons, 1) | |||
| 650 | + | require.Len(t, reopened.Daemons[0].Sessions, 1) | |||
| 651 | + | for _, tab := range reopened.Daemons[0].Sessions[0].Tabs { | |||
| 652 | + | require.False(t, tab.Attention) | |||
| 653 | + | } | |||
| 654 | + | }) | |||
| 655 | + | } | |||
| 656 | + | } | |||
| 657 | + | ||||
| 578 | func TestFencingAndCommitFaults(t *testing.T) { | 658 | func TestFencingAndCommitFaults(t *testing.T) { | ||
| 579 | for _, point := range []string{"write", "sync", "rename", "dirsync"} { | 659 | for _, point := range []string{"write", "sync", "rename", "dirsync"} { | ||
| 580 | t.Run(point, func(t *testing.T) { | 660 | t.Run(point, func(t *testing.T) { | ||