From 2d592c63c16ca63a3dd97a1cda656bfc8e5d564f Mon Sep 17 00:00:00 2001 From: jochens Date: Fri, 13 Nov 2026 01:45:19 +0100 Subject: [PATCH] zfs: a keeper that does not answer is warned once, and RunKeeper closes its second connection (review round 2 finding 2, A-37) A failed keeper call in a root process (hand acts, forget, Paused, every call) now adds one warning that the bus is taken as down, instead of being dropped silently. RunKeeper closes the connection acct.Dial opens when the keeper stops; the message id's comment says it is nanotime-pid. Builds: design 42 zfs-owns-zfs A-37 --- modules/zfs/cmd/zfs-tools/bus.go | 13 ++++++++--- modules/zfs/cmd/zfs-tools/keeper.go | 26 ++++++++++++++++------ modules/zfs/cmd/zfs-tools/keeper_test.go | 28 +++++++++++++++++++++++- 3 files changed, 56 insertions(+), 11 deletions(-) diff --git a/modules/zfs/cmd/zfs-tools/bus.go b/modules/zfs/cmd/zfs-tools/bus.go index af6ffb6c5..b5da08576 100644 --- a/modules/zfs/cmd/zfs-tools/bus.go +++ b/modules/zfs/cmd/zfs-tools/bus.go @@ -48,7 +48,13 @@ func (m *Machine) JoinBus() func() { } // Every call reaches the socket anew, so a keeper that comes up after this process is used from then on; one that // is down at a call answers that call with an error, never the bus taken as down for the process's life. - c := keeperClient{sock} + // A call the keeper does not answer is said once, as a warning: the root process then acts as with the bus down (A-37). + var once sync.Once + c := keeperClient{sock: sock, down: func(err error) { + once.Do(func() { + m.Warnings = append(m.Warnings, "the bus is taken as down: "+err.Error()) + }) + }} m.Emit = c.emit m.States = c m.Findings = meshconditionsclient.FindingWriter{State: findingState{c}} @@ -75,10 +81,11 @@ func RunKeeper(ctx context.Context) error { if err != nil { return err } - _, js, err := acct.Dial() + nc, js, err := acct.Dial() if err != nil { return err } + defer nc.Close() k := &Keeper{Emit: func(kind string, body any) error { return publishEvent(js, acct, kind, body) }, States: busStates{c}, Call: c.Call} return k.Serve(ctx, l) } @@ -90,7 +97,7 @@ type eventPublisher interface { // publishEvent puts one event on the module's events stream as a JetStream publish the bus acknowledges, on the // subject the stream captures, mesh.bus..events. (A-41); never process.Conn.Emit, whose subject the -// keeper's account is not granted. The message id is a random one, as before. +// keeper's account is not granted. The message id (nanotime-pid) is the time in nanoseconds and the process id, so two events of one process never share one. func publishEvent(js eventPublisher, acct process.Account, kind string, body any) error { data, err := json.Marshal(body) if err != nil { diff --git a/modules/zfs/cmd/zfs-tools/keeper.go b/modules/zfs/cmd/zfs-tools/keeper.go index fc63abf7e..0ffa12d79 100644 --- a/modules/zfs/cmd/zfs-tools/keeper.go +++ b/modules/zfs/cmd/zfs-tools/keeper.go @@ -239,26 +239,38 @@ func Listen(sock string) (net.Listener, error) { } // keeperClient is a root process's way to the bus: calls on the keeper's socket. -type keeperClient struct{ sock string } +type keeperClient struct { + sock string + // down, when set, is told of a call the keeper did not answer (not of an answer that is an error). + down func(error) +} func (c keeperClient) do(r KeeperRequest) (KeeperAnswer, error) { + a, err, unanswered := c.exchange(r) + if unanswered && c.down != nil { + c.down(err) + } + return a, err +} + +// exchange is one call on the socket; unanswered says the keeper gave no answer (as opposed to an answer that is an error). +func (c keeperClient) exchange(r KeeperRequest) (a KeeperAnswer, err error, unanswered bool) { conn, err := net.DialTimeout("unix", c.sock, 5*time.Second) if err != nil { - return KeeperAnswer{}, fmt.Errorf("the keeper does not answer on %s: %v", c.sock, err) + return KeeperAnswer{}, fmt.Errorf("the keeper does not answer on %s: %v", c.sock, err), true } defer conn.Close() _ = conn.SetDeadline(time.Now().Add(KeeperTimeout)) if err := json.NewEncoder(conn).Encode(r); err != nil { - return KeeperAnswer{}, err + return KeeperAnswer{}, err, true } - var a KeeperAnswer if err := json.NewDecoder(conn).Decode(&a); err != nil { - return KeeperAnswer{}, fmt.Errorf("the keeper's answer: %v", err) + return KeeperAnswer{}, fmt.Errorf("the keeper's answer: %v", err), true } if a.Error != "" { - return a, errors.New(a.Error) + return a, errors.New(a.Error), false } - return a, nil + return a, nil, false } func (c keeperClient) emit(kind string, body any) error { diff --git a/modules/zfs/cmd/zfs-tools/keeper_test.go b/modules/zfs/cmd/zfs-tools/keeper_test.go index fda12c202..c27ee3e72 100644 --- a/modules/zfs/cmd/zfs-tools/keeper_test.go +++ b/modules/zfs/cmd/zfs-tools/keeper_test.go @@ -86,7 +86,7 @@ func TestOnlyRootOpensTheKeepersSocketAndANeedlessKeeperIsTheBusDown(t *testing. ctx, stop := context.WithCancel(context.Background()) defer stop() go k.Serve(ctx, l) - c := keeperClient{sock} + c := keeperClient{sock: sock} if err := c.Put(StateSchedule, "snapshots", map[string]string{"by": "x"}); err != nil { t.Fatal(err) } @@ -144,3 +144,29 @@ func TestOnlyRootOpensTheKeepersSocketAndANeedlessKeeperIsTheBusDown(t *testing. t.Errorf("a keeper answering: the bus is up") } } + +// A-37: a root process whose keeper does not answer acts as with the bus down and says so, once, in a warning, for the +// hand acts, the forgetting of applied keys and the pause alike. +func TestAKeeperThatDoesNotAnswerIsWarnedOnce(t *testing.T) { + dir, err := os.MkdirTemp("/var/tmp", "keeper-down") + if err != nil { + t.Fatal(err) + } + defer os.RemoveAll(dir) + t.Setenv(KeeperSocketEnv, filepath.Join(dir, "none.sock")) + m := newFake(t).machine() + m.JoinBus() + if got := m.handActs(); got != nil { + t.Errorf("hand acts with no keeper: %v", got) + } + if len(m.Warnings) != 1 || !strings.Contains(m.Warnings[0], "bus is taken as down") { + t.Fatalf("hand acts: %v", m.Warnings) + } + m.forget(&Config{}) + if m.Paused() { + t.Errorf("paused with no keeper") + } + if len(m.Warnings) != 1 { + t.Errorf("said more than once: %v", m.Warnings) + } +}