fix(broker): keep list and host reads truthful without spawning
- commit
- d031a8e3d14fd986d8c67fafbd169a290a663034
- author
- brice · · 2026-09-23 09:28 UTC
- parent
- b71082c
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
9 files · +281 -16
M internal/adapters/brokerstore/store.go +17 -0
| @@ -117,6 +117,23 @@ type Store struct { | |||
| 117 | 117 | ||
| 118 | 118 | var _ ports.BrokerHostStore = (*Store)(nil) | |
| 119 | 119 | ||
| 120 | + | // ReadHosts reads committed membership without taking the writer lock or | |
| 121 | + | // creating any files. Advisory snapshots are deliberately ignored. | |
| 122 | + | func ReadHosts(dir string) (ports.BrokerHosts, error) { | |
| 123 | + | raw, err := readBounded(filepath.Join(dir, "state.json")) | |
| 124 | + | if err != nil { | |
| 125 | + | return ports.BrokerHosts{}, err | |
| 126 | + | } | |
| 127 | + | var st state | |
| 128 | + | if err := strict(raw, &st); err != nil { | |
| 129 | + | return ports.BrokerHosts{}, invalid("state.json: %v", err) | |
| 130 | + | } | |
| 131 | + | if err := validate(st); err != nil { | |
| 132 | + | return ports.BrokerHosts{}, err | |
| 133 | + | } | |
| 134 | + | return ports.BrokerHosts{Revision: st.Hosts.Revision, Hosts: ports.CloneBrokerHostRecords(st.Hosts.Hosts)}, nil | |
| 135 | + | } | |
| 136 | + | ||
| 120 | 137 | // Open returns only after validated unified state is durable. The | |
| 121 | 138 | // recovery record is written first and retained forever as the rollback copy; | |
| 122 | 139 | // restart reuses its generated identities instead of migrating a second time. | |
| @@ -117,6 +117,23 @@ type Store struct { | |||||
| 117 | 117 | ||||
| 118 | var _ ports.BrokerHostStore = (*Store)(nil) | 118 | var _ ports.BrokerHostStore = (*Store)(nil) | ||
| 119 | 119 | ||||
| 120 | + | // ReadHosts reads committed membership without taking the writer lock or | |||
| 121 | + | // creating any files. Advisory snapshots are deliberately ignored. | |||
| 122 | + | func ReadHosts(dir string) (ports.BrokerHosts, error) { | |||
| 123 | + | raw, err := readBounded(filepath.Join(dir, "state.json")) | |||
| 124 | + | if err != nil { | |||
| 125 | + | return ports.BrokerHosts{}, err | |||
| 126 | + | } | |||
| 127 | + | var st state | |||
| 128 | + | if err := strict(raw, &st); err != nil { | |||
| 129 | + | return ports.BrokerHosts{}, invalid("state.json: %v", err) | |||
| 130 | + | } | |||
| 131 | + | if err := validate(st); err != nil { | |||
| 132 | + | return ports.BrokerHosts{}, err | |||
| 133 | + | } | |||
| 134 | + | return ports.BrokerHosts{Revision: st.Hosts.Revision, Hosts: ports.CloneBrokerHostRecords(st.Hosts.Hosts)}, nil | |||
| 135 | + | } | |||
| 136 | + | ||||
| 120 | // Open returns only after validated unified state is durable. The | 137 | // Open returns only after validated unified state is durable. The | ||
| 121 | // recovery record is written first and retained forever as the rollback copy; | 138 | // recovery record is written first and retained forever as the rollback copy; | ||
| 122 | // restart reuses its generated identities instead of migrating a second time. | 139 | // restart reuses its generated identities instead of migrating a second time. | ||
M internal/app/broker_daemon_start.go +3 -0
| @@ -156,6 +156,9 @@ func dialBrokerDaemonCarriage(ctx context.Context, carriage string, target ports | |||
| 156 | 156 | if target.StartMode == ports.BrokerDaemonExistingOnly { | |
| 157 | 157 | // An existing-only acquisition is one dial: no lifecycle probe, no spawn | |
| 158 | 158 | // lock, no process, and no fallback of any kind. | |
| 159 | + | if daemonCarriageAbsence(err) { | |
| 160 | + | return nil, ports.BrokerError{Code: ports.BrokerErrorNoDaemon, Cause: err} | |
| 161 | + | } | |
| 159 | 162 | return nil, err | |
| 160 | 163 | } | |
| 161 | 164 | // A refusal that spawning could never repair is reported unchanged: a | |
| @@ -156,6 +156,9 @@ func dialBrokerDaemonCarriage(ctx context.Context, carriage string, target ports | |||||
| 156 | if target.StartMode == ports.BrokerDaemonExistingOnly { | 156 | if target.StartMode == ports.BrokerDaemonExistingOnly { | ||
| 157 | // An existing-only acquisition is one dial: no lifecycle probe, no spawn | 157 | // An existing-only acquisition is one dial: no lifecycle probe, no spawn | ||
| 158 | // lock, no process, and no fallback of any kind. | 158 | // lock, no process, and no fallback of any kind. | ||
| 159 | + | if daemonCarriageAbsence(err) { | |||
| 160 | + | return nil, ports.BrokerError{Code: ports.BrokerErrorNoDaemon, Cause: err} | |||
| 161 | + | } | |||
| 159 | return nil, err | 162 | return nil, err | ||
| 160 | } | 163 | } | ||
| 161 | // A refusal that spawning could never repair is reported unchanged: a | 164 | // A refusal that spawning could never repair is reported unchanged: a | ||
M internal/app/broker_daemon_start_test.go +3 -0
| @@ -250,6 +250,9 @@ func TestDaemonStartExistingOnlyNeverSpawns(t *testing.T) { | |||
| 250 | 250 | target := startableLocalTarget(t, carriage, ports.BrokerDaemonExistingOnly) | |
| 251 | 251 | raw, err := dialBrokerLocalDaemon(context.Background(), carriage, target) | |
| 252 | 252 | require.ErrorIs(t, err, os.ErrNotExist) | |
| 253 | + | var brokerErr ports.BrokerError | |
| 254 | + | require.ErrorAs(t, err, &brokerErr) | |
| 255 | + | require.Equal(t, ports.BrokerErrorNoDaemon, brokerErr.Code) | |
| 253 | 256 | require.Nil(t, raw) | |
| 254 | 257 | ||
| 255 | 258 | dials, spawns := scripted.counters() | |
| @@ -250,6 +250,9 @@ func TestDaemonStartExistingOnlyNeverSpawns(t *testing.T) { | |||||
| 250 | target := startableLocalTarget(t, carriage, ports.BrokerDaemonExistingOnly) | 250 | target := startableLocalTarget(t, carriage, ports.BrokerDaemonExistingOnly) | ||
| 251 | raw, err := dialBrokerLocalDaemon(context.Background(), carriage, target) | 251 | raw, err := dialBrokerLocalDaemon(context.Background(), carriage, target) | ||
| 252 | require.ErrorIs(t, err, os.ErrNotExist) | 252 | require.ErrorIs(t, err, os.ErrNotExist) | ||
| 253 | + | var brokerErr ports.BrokerError | |||
| 254 | + | require.ErrorAs(t, err, &brokerErr) | |||
| 255 | + | require.Equal(t, ports.BrokerErrorNoDaemon, brokerErr.Code) | |||
| 253 | require.Nil(t, raw) | 256 | require.Nil(t, raw) | ||
| 254 | 257 | ||||
| 255 | dials, spawns := scripted.counters() | 258 | dials, spawns := scripted.counters() | ||
A internal/app/list_readonly_test.go +109 -0
| @@ -0,0 +1,109 @@ | |||
| 1 | + | package app | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "bytes" | |
| 5 | + | "context" | |
| 6 | + | "errors" | |
| 7 | + | "os" | |
| 8 | + | "path/filepath" | |
| 9 | + | "testing" | |
| 10 | + | "time" | |
| 11 | + | ||
| 12 | + | "github.com/stretchr/testify/require" | |
| 13 | + | ||
| 14 | + | "github.com/bnema/vev/internal/domain" | |
| 15 | + | "github.com/bnema/vev/internal/ports" | |
| 16 | + | "github.com/bnema/vev/internal/protocol/catalogue" | |
| 17 | + | ) | |
| 18 | + | ||
| 19 | + | func TestObservedHostStatus(t *testing.T) { | |
| 20 | + | now := time.Date(2026, 9, 23, 9, 0, 0, 0, time.UTC) | |
| 21 | + | for _, tc := range []struct { | |
| 22 | + | name string | |
| 23 | + | availability domain.RemoteAvailability | |
| 24 | + | age time.Duration | |
| 25 | + | want string | |
| 26 | + | show bool | |
| 27 | + | }{ | |
| 28 | + | {"fresh", domain.RemoteAvailabilityReachable, time.Second, "reachable", true}, | |
| 29 | + | {"stale", domain.RemoteAvailabilityReachable, 2 * time.Minute, "stale (2m0s since last observation)", false}, | |
| 30 | + | {"unknown success", domain.RemoteAvailabilityReachable, 0, "stale (last observation unknown)", false}, | |
| 31 | + | {"never observed", domain.RemoteAvailabilityUnknown, time.Second, "not yet observed", false}, | |
| 32 | + | {"down", domain.RemoteAvailabilityUnreachable, time.Second, "down (unreachable)", false}, | |
| 33 | + | {"auth", domain.RemoteAvailabilityAuthFailed, time.Second, "down (authentication_failed)", false}, | |
| 34 | + | {"invalid", domain.RemoteAvailabilityInvalidResponse, time.Second, "down (invalid_response)", false}, | |
| 35 | + | {"incompatible", domain.RemoteAvailabilityIncompatible, time.Second, "incompatible (identity changed? use vev host rm/add)", false}, | |
| 36 | + | {"no daemon", domain.RemoteAvailabilityNoDaemon, time.Second, "no vev daemon — Enter in the picker to create a session", false}, | |
| 37 | + | } { | |
| 38 | + | t.Run(tc.name, func(t *testing.T) { | |
| 39 | + | observed := ports.BrokerDaemonObservation{Availability: tc.availability} | |
| 40 | + | if tc.age != 0 { | |
| 41 | + | observed.LastSuccess = now.Add(-tc.age) | |
| 42 | + | } | |
| 43 | + | status, show := observedHostStatus(observed, now) | |
| 44 | + | require.Equal(t, tc.want, status) | |
| 45 | + | require.Equal(t, tc.show, show) | |
| 46 | + | }) | |
| 47 | + | } | |
| 48 | + | } | |
| 49 | + | ||
| 50 | + | func TestSnapshotListNeverShowsCachedUpWhenHostIsNotFresh(t *testing.T) { | |
| 51 | + | now := time.Date(2026, 9, 23, 9, 0, 0, 0, time.UTC) | |
| 52 | + | for _, availability := range []domain.RemoteAvailability{domain.RemoteAvailabilityReachable, domain.RemoteAvailabilityUnreachable, domain.RemoteAvailabilityNoDaemon} { | |
| 53 | + | t.Run(availability.String(), func(t *testing.T) { | |
| 54 | + | remote := ports.BrokerDaemonObservation{ | |
| 55 | + | Endpoint: "user@example.test", DisplayOrigin: "user@example.test", | |
| 56 | + | Registration: domain.RemoteRegistration{Endpoint: "user@example.test", Incarnation: [16]byte{1}, Generation: 1}, | |
| 57 | + | Policy: seamBrokerTestPolicy(), Availability: availability, | |
| 58 | + | LastSuccess: now.Add(-time.Minute), InventoryKnown: true, | |
| 59 | + | Sessions: []catalogue.RemoteCatalogSession{{LifecycleID: domain.SessionLifecycleID{1}, Name: "old", State: catalogue.RemoteCatalogSessionUp}}, | |
| 60 | + | } | |
| 61 | + | snapshot := ports.BrokerSnapshot{Epoch: seamBrokerTestEpoch, Revision: 1, Daemons: []ports.BrokerDaemonObservation{remote}} | |
| 62 | + | var output bytes.Buffer | |
| 63 | + | require.NoError(t, renderBrokerSnapshotList(command{listAll: true}, snapshot, &output, now)) | |
| 64 | + | require.NotContains(t, output.String(), "old@") | |
| 65 | + | require.NotContains(t, output.String(), " up") | |
| 66 | + | }) | |
| 67 | + | } | |
| 68 | + | } | |
| 69 | + | ||
| 70 | + | func TestRunListWithoutBrokerDoesNotStartEitherProcess(t *testing.T) { | |
| 71 | + | previous := connectListBroker | |
| 72 | + | connectListBroker = func(context.Context) (ports.BrokerService, error) { return nil, errBrokerAbsent } | |
| 73 | + | t.Cleanup(func() { connectListBroker = previous }) | |
| 74 | + | root := t.TempDir() | |
| 75 | + | t.Setenv("XDG_RUNTIME_DIR", filepath.Join(root, "runtime")) | |
| 76 | + | t.Setenv("XDG_STATE_HOME", filepath.Join(root, "state")) | |
| 77 | + | t.Setenv("XDG_CONFIG_HOME", filepath.Join(root, "config")) | |
| 78 | + | for _, tc := range []struct { | |
| 79 | + | name string | |
| 80 | + | cmd command | |
| 81 | + | want string | |
| 82 | + | }{ | |
| 83 | + | {"local", command{kind: kindList}, "no sessions"}, | |
| 84 | + | {"remote", command{kind: kindList, listHost: "user@example.test"}, "not connected (no broker running)"}, | |
| 85 | + | {"all", command{kind: kindList, listAll: true}, "no hosts"}, | |
| 86 | + | } { | |
| 87 | + | t.Run(tc.name, func(t *testing.T) { | |
| 88 | + | output := captureStdout(t, func() { require.NoError(t, runList(context.Background(), tc.cmd)) }) | |
| 89 | + | require.Contains(t, output, tc.want) | |
| 90 | + | }) | |
| 91 | + | } | |
| 92 | + | _, err := os.Stat(filepath.Join(root, "runtime", "vev")) | |
| 93 | + | require.True(t, errors.Is(err, os.ErrNotExist)) | |
| 94 | + | } | |
| 95 | + | ||
| 96 | + | func TestHostListUsesExistingConnector(t *testing.T) { | |
| 97 | + | var output bytes.Buffer | |
| 98 | + | called := false | |
| 99 | + | deps := remoteHostDeps{ | |
| 100 | + | connect: func(context.Context) (ports.BrokerService, error) { t.Fatal("list spawned a broker"); return nil, nil }, | |
| 101 | + | connectExisting: func(context.Context) (ports.BrokerService, error) { called = true; return nil, errBrokerAbsent }, | |
| 102 | + | stdout: &output, | |
| 103 | + | } | |
| 104 | + | t.Setenv("XDG_STATE_HOME", t.TempDir()) | |
| 105 | + | t.Setenv("XDG_CONFIG_HOME", t.TempDir()) | |
| 106 | + | require.NoError(t, runHostCommand(context.Background(), command{hostAction: hostActionList}, deps)) | |
| 107 | + | require.True(t, called) | |
| 108 | + | require.Equal(t, "no hosts\n", output.String()) | |
| 109 | + | } | |
| @@ -0,0 +1,109 @@ | |||||
| 1 | + | package app | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "bytes" | |||
| 5 | + | "context" | |||
| 6 | + | "errors" | |||
| 7 | + | "os" | |||
| 8 | + | "path/filepath" | |||
| 9 | + | "testing" | |||
| 10 | + | "time" | |||
| 11 | + | ||||
| 12 | + | "github.com/stretchr/testify/require" | |||
| 13 | + | ||||
| 14 | + | "github.com/bnema/vev/internal/domain" | |||
| 15 | + | "github.com/bnema/vev/internal/ports" | |||
| 16 | + | "github.com/bnema/vev/internal/protocol/catalogue" | |||
| 17 | + | ) | |||
| 18 | + | ||||
| 19 | + | func TestObservedHostStatus(t *testing.T) { | |||
| 20 | + | now := time.Date(2026, 9, 23, 9, 0, 0, 0, time.UTC) | |||
| 21 | + | for _, tc := range []struct { | |||
| 22 | + | name string | |||
| 23 | + | availability domain.RemoteAvailability | |||
| 24 | + | age time.Duration | |||
| 25 | + | want string | |||
| 26 | + | show bool | |||
| 27 | + | }{ | |||
| 28 | + | {"fresh", domain.RemoteAvailabilityReachable, time.Second, "reachable", true}, | |||
| 29 | + | {"stale", domain.RemoteAvailabilityReachable, 2 * time.Minute, "stale (2m0s since last observation)", false}, | |||
| 30 | + | {"unknown success", domain.RemoteAvailabilityReachable, 0, "stale (last observation unknown)", false}, | |||
| 31 | + | {"never observed", domain.RemoteAvailabilityUnknown, time.Second, "not yet observed", false}, | |||
| 32 | + | {"down", domain.RemoteAvailabilityUnreachable, time.Second, "down (unreachable)", false}, | |||
| 33 | + | {"auth", domain.RemoteAvailabilityAuthFailed, time.Second, "down (authentication_failed)", false}, | |||
| 34 | + | {"invalid", domain.RemoteAvailabilityInvalidResponse, time.Second, "down (invalid_response)", false}, | |||
| 35 | + | {"incompatible", domain.RemoteAvailabilityIncompatible, time.Second, "incompatible (identity changed? use vev host rm/add)", false}, | |||
| 36 | + | {"no daemon", domain.RemoteAvailabilityNoDaemon, time.Second, "no vev daemon — Enter in the picker to create a session", false}, | |||
| 37 | + | } { | |||
| 38 | + | t.Run(tc.name, func(t *testing.T) { | |||
| 39 | + | observed := ports.BrokerDaemonObservation{Availability: tc.availability} | |||
| 40 | + | if tc.age != 0 { | |||
| 41 | + | observed.LastSuccess = now.Add(-tc.age) | |||
| 42 | + | } | |||
| 43 | + | status, show := observedHostStatus(observed, now) | |||
| 44 | + | require.Equal(t, tc.want, status) | |||
| 45 | + | require.Equal(t, tc.show, show) | |||
| 46 | + | }) | |||
| 47 | + | } | |||
| 48 | + | } | |||
| 49 | + | ||||
| 50 | + | func TestSnapshotListNeverShowsCachedUpWhenHostIsNotFresh(t *testing.T) { | |||
| 51 | + | now := time.Date(2026, 9, 23, 9, 0, 0, 0, time.UTC) | |||
| 52 | + | for _, availability := range []domain.RemoteAvailability{domain.RemoteAvailabilityReachable, domain.RemoteAvailabilityUnreachable, domain.RemoteAvailabilityNoDaemon} { | |||
| 53 | + | t.Run(availability.String(), func(t *testing.T) { | |||
| 54 | + | remote := ports.BrokerDaemonObservation{ | |||
| 55 | + | Endpoint: "user@example.test", DisplayOrigin: "user@example.test", | |||
| 56 | + | Registration: domain.RemoteRegistration{Endpoint: "user@example.test", Incarnation: [16]byte{1}, Generation: 1}, | |||
| 57 | + | Policy: seamBrokerTestPolicy(), Availability: availability, | |||
| 58 | + | LastSuccess: now.Add(-time.Minute), InventoryKnown: true, | |||
| 59 | + | Sessions: []catalogue.RemoteCatalogSession{{LifecycleID: domain.SessionLifecycleID{1}, Name: "old", State: catalogue.RemoteCatalogSessionUp}}, | |||
| 60 | + | } | |||
| 61 | + | snapshot := ports.BrokerSnapshot{Epoch: seamBrokerTestEpoch, Revision: 1, Daemons: []ports.BrokerDaemonObservation{remote}} | |||
| 62 | + | var output bytes.Buffer | |||
| 63 | + | require.NoError(t, renderBrokerSnapshotList(command{listAll: true}, snapshot, &output, now)) | |||
| 64 | + | require.NotContains(t, output.String(), "old@") | |||
| 65 | + | require.NotContains(t, output.String(), " up") | |||
| 66 | + | }) | |||
| 67 | + | } | |||
| 68 | + | } | |||
| 69 | + | ||||
| 70 | + | func TestRunListWithoutBrokerDoesNotStartEitherProcess(t *testing.T) { | |||
| 71 | + | previous := connectListBroker | |||
| 72 | + | connectListBroker = func(context.Context) (ports.BrokerService, error) { return nil, errBrokerAbsent } | |||
| 73 | + | t.Cleanup(func() { connectListBroker = previous }) | |||
| 74 | + | root := t.TempDir() | |||
| 75 | + | t.Setenv("XDG_RUNTIME_DIR", filepath.Join(root, "runtime")) | |||
| 76 | + | t.Setenv("XDG_STATE_HOME", filepath.Join(root, "state")) | |||
| 77 | + | t.Setenv("XDG_CONFIG_HOME", filepath.Join(root, "config")) | |||
| 78 | + | for _, tc := range []struct { | |||
| 79 | + | name string | |||
| 80 | + | cmd command | |||
| 81 | + | want string | |||
| 82 | + | }{ | |||
| 83 | + | {"local", command{kind: kindList}, "no sessions"}, | |||
| 84 | + | {"remote", command{kind: kindList, listHost: "user@example.test"}, "not connected (no broker running)"}, | |||
| 85 | + | {"all", command{kind: kindList, listAll: true}, "no hosts"}, | |||
| 86 | + | } { | |||
| 87 | + | t.Run(tc.name, func(t *testing.T) { | |||
| 88 | + | output := captureStdout(t, func() { require.NoError(t, runList(context.Background(), tc.cmd)) }) | |||
| 89 | + | require.Contains(t, output, tc.want) | |||
| 90 | + | }) | |||
| 91 | + | } | |||
| 92 | + | _, err := os.Stat(filepath.Join(root, "runtime", "vev")) | |||
| 93 | + | require.True(t, errors.Is(err, os.ErrNotExist)) | |||
| 94 | + | } | |||
| 95 | + | ||||
| 96 | + | func TestHostListUsesExistingConnector(t *testing.T) { | |||
| 97 | + | var output bytes.Buffer | |||
| 98 | + | called := false | |||
| 99 | + | deps := remoteHostDeps{ | |||
| 100 | + | connect: func(context.Context) (ports.BrokerService, error) { t.Fatal("list spawned a broker"); return nil, nil }, | |||
| 101 | + | connectExisting: func(context.Context) (ports.BrokerService, error) { called = true; return nil, errBrokerAbsent }, | |||
| 102 | + | stdout: &output, | |||
| 103 | + | } | |||
| 104 | + | t.Setenv("XDG_STATE_HOME", t.TempDir()) | |||
| 105 | + | t.Setenv("XDG_CONFIG_HOME", t.TempDir()) | |||
| 106 | + | require.NoError(t, runHostCommand(context.Background(), command{hostAction: hostActionList}, deps)) | |||
| 107 | + | require.True(t, called) | |||
| 108 | + | require.Equal(t, "no hosts\n", output.String()) | |||
| 109 | + | } | |||
M internal/app/remote_hosts.go +94 -9
| @@ -2,12 +2,16 @@ package app | |||
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | 4 | "context" | |
| 5 | + | "errors" | |
| 5 | 6 | "fmt" | |
| 6 | 7 | "io" | |
| 7 | 8 | "os" | |
| 8 | 9 | "sort" | |
| 9 | 10 | "text/tabwriter" | |
| 11 | + | "time" | |
| 10 | 12 | ||
| 13 | + | "github.com/bnema/vev/internal/adapters/brokerconfig" | |
| 14 | + | "github.com/bnema/vev/internal/adapters/brokerstore" | |
| 11 | 15 | "github.com/bnema/vev/internal/domain" | |
| 12 | 16 | "github.com/bnema/vev/internal/ports" | |
| 13 | 17 | "github.com/bnema/vev/internal/protocol" | |
| @@ -21,18 +25,22 @@ const ( | |||
| 21 | 25 | ) | |
| 22 | 26 | ||
| 23 | 27 | type remoteHostDeps struct { | |
| 24 | - | connect func(context.Context) (ports.BrokerService, error) | |
| 25 | - | stdout io.Writer | |
| 28 | + | connect func(context.Context) (ports.BrokerService, error) | |
| 29 | + | connectExisting func(context.Context) (ports.BrokerService, error) | |
| 30 | + | stdout io.Writer | |
| 26 | 31 | } | |
| 27 | 32 | ||
| 28 | 33 | func defaultRemoteHostDeps() remoteHostDeps { | |
| 29 | - | return remoteHostDeps{connect: connectProductionBroker, stdout: os.Stdout} | |
| 34 | + | return remoteHostDeps{connect: connectProductionBroker, connectExisting: connectExistingBroker, stdout: os.Stdout} | |
| 30 | 35 | } | |
| 31 | 36 | ||
| 32 | 37 | func (d remoteHostDeps) withDefaults() remoteHostDeps { | |
| 33 | 38 | if d.connect == nil { | |
| 34 | 39 | d.connect = connectProductionBroker | |
| 35 | 40 | } | |
| 41 | + | if d.connectExisting == nil { | |
| 42 | + | d.connectExisting = connectExistingBroker | |
| 43 | + | } | |
| 36 | 44 | if d.stdout == nil { | |
| 37 | 45 | d.stdout = os.Stdout | |
| 38 | 46 | } | |
| @@ -41,8 +49,15 @@ func (d remoteHostDeps) withDefaults() remoteHostDeps { | |||
| 41 | 49 | ||
| 42 | 50 | func runHostCommand(ctx context.Context, cmd command, deps remoteHostDeps) error { | |
| 43 | 51 | deps = deps.withDefaults() | |
| 44 | - | service, err := deps.connect(ctx) | |
| 52 | + | connect := deps.connect | |
| 53 | + | if cmd.hostAction == hostActionList { | |
| 54 | + | connect = deps.connectExisting | |
| 55 | + | } | |
| 56 | + | service, err := connect(ctx) | |
| 45 | 57 | if err != nil { | |
| 58 | + | if cmd.hostAction == hostActionList && errors.Is(err, errBrokerAbsent) { | |
| 59 | + | return listDisconnectedHosts(deps.stdout) | |
| 60 | + | } | |
| 46 | 61 | return err | |
| 47 | 62 | } | |
| 48 | 63 | defer func() { _ = service.Close() }() | |
| @@ -86,6 +101,42 @@ func runHostCommand(ctx context.Context, cmd command, deps remoteHostDeps) error | |||
| 86 | 101 | } | |
| 87 | 102 | } | |
| 88 | 103 | ||
| 104 | + | // listDisconnectedHosts reads configured endpoints without dialing any remote. | |
| 105 | + | func listDisconnectedHosts(w io.Writer) error { | |
| 106 | + | state := productionBrokerLayout().State | |
| 107 | + | hosts, err := brokerstore.ReadHosts(state) | |
| 108 | + | var endpoints []string | |
| 109 | + | if errors.Is(err, os.ErrNotExist) { | |
| 110 | + | // Before the first broker run, the config is the only membership. | |
| 111 | + | config, configErr := brokerconfig.LoadPath(productionBrokerLayout(), productionBrokerConfigPath()) | |
| 112 | + | if errors.Is(configErr, os.ErrNotExist) { | |
| 113 | + | _, err = fmt.Fprintln(w, "no hosts") | |
| 114 | + | return err | |
| 115 | + | } | |
| 116 | + | if configErr != nil { | |
| 117 | + | return configErr | |
| 118 | + | } | |
| 119 | + | endpoints = config.Endpoints() | |
| 120 | + | } else if err != nil { | |
| 121 | + | return err | |
| 122 | + | } else { | |
| 123 | + | for _, host := range hosts.Hosts { | |
| 124 | + | endpoints = append(endpoints, host.Registration.Endpoint) | |
| 125 | + | } | |
| 126 | + | sort.Strings(endpoints) | |
| 127 | + | } | |
| 128 | + | if len(endpoints) == 0 { | |
| 129 | + | _, err = fmt.Fprintln(w, "no hosts") | |
| 130 | + | return err | |
| 131 | + | } | |
| 132 | + | for _, endpoint := range endpoints { | |
| 133 | + | if _, err := fmt.Fprintf(w, "%s: not connected (no broker running)\n", endpoint); err != nil { | |
| 134 | + | return err | |
| 135 | + | } | |
| 136 | + | } | |
| 137 | + | return nil | |
| 138 | + | } | |
| 139 | + | ||
| 89 | 140 | func printBrokerHosts(w io.Writer, snapshot ports.BrokerSnapshot) error { | |
| 90 | 141 | if err := snapshot.Validate(); err != nil { | |
| 91 | 142 | return fmt.Errorf("vev: reading broker catalogue: %w", err) | |
| @@ -104,10 +155,10 @@ func printBrokerHosts(w io.Writer, snapshot ports.BrokerSnapshot) error { | |||
| 104 | 155 | tw := tabwriter.NewWriter(w, 0, 4, 2, ' ', 0) | |
| 105 | 156 | _, _ = fmt.Fprintln(tw, "TARGET\tSOURCE\tSTATUS") | |
| 106 | 157 | for _, endpoint := range endpoints { | |
| 107 | - | status := "configured" | |
| 158 | + | status := "not connected" | |
| 108 | 159 | for _, daemon := range snapshot.Daemons { | |
| 109 | - | if !daemon.Local && daemon.Endpoint == endpoint && daemon.Availability == domain.RemoteAvailabilityNoDaemon { | |
| 110 | - | status = "no daemon" | |
| 160 | + | if !daemon.Local && daemon.Endpoint == endpoint { | |
| 161 | + | status, _ = observedHostStatus(daemon, time.Now()) | |
| 111 | 162 | break | |
| 112 | 163 | } | |
| 113 | 164 | } | |
| @@ -117,6 +168,10 @@ func printBrokerHosts(w io.Writer, snapshot ports.BrokerSnapshot) error { | |||
| 117 | 168 | } | |
| 118 | 169 | ||
| 119 | 170 | func runBrokerSnapshotList(cmd command, snapshot ports.BrokerSnapshot, stdout io.Writer) error { | |
| 171 | + | return renderBrokerSnapshotList(cmd, snapshot, stdout, time.Now()) | |
| 172 | + | } | |
| 173 | + | ||
| 174 | + | func renderBrokerSnapshotList(cmd command, snapshot ports.BrokerSnapshot, stdout io.Writer, now time.Time) error { | |
| 120 | 175 | if err := snapshot.Validate(); err != nil { | |
| 121 | 176 | return fmt.Errorf("vev: reading broker catalogue: %w", err) | |
| 122 | 177 | } | |
| @@ -128,8 +183,15 @@ func runBrokerSnapshotList(cmd command, snapshot ports.BrokerSnapshot, stdout io | |||
| 128 | 183 | continue | |
| 129 | 184 | } | |
| 130 | 185 | found = true | |
| 131 | - | if !daemon.Local && daemon.Availability == domain.RemoteAvailabilityNoDaemon { | |
| 132 | - | statuses = append(statuses, fmt.Sprintf("%s: no vev daemon — Enter in the picker to create a session", daemon.Endpoint)) | |
| 186 | + | status, showSessions := observedHostStatus(daemon, now) | |
| 187 | + | if daemon.Local { | |
| 188 | + | showSessions = daemon.Availability == domain.RemoteAvailabilityReachable | |
| 189 | + | } | |
| 190 | + | if !daemon.Local && status != "reachable" { | |
| 191 | + | statuses = append(statuses, fmt.Sprintf("%s: %s", daemon.Endpoint, status)) | |
| 192 | + | } | |
| 193 | + | if !showSessions { | |
| 194 | + | continue | |
| 133 | 195 | } | |
| 134 | 196 | for i, session := range catalogSessionsAsInfo(daemon.Endpoint, daemon.Sessions) { | |
| 135 | 197 | if daemon.Local { | |
| @@ -153,6 +215,29 @@ func runBrokerSnapshotList(cmd command, snapshot ports.BrokerSnapshot, stdout io | |||
| 153 | 215 | return nil | |
| 154 | 216 | } | |
| 155 | 217 | ||
| 218 | + | // observedHostStatus never treats cached inventory as proof of current reachability. | |
| 219 | + | func observedHostStatus(daemon ports.BrokerDaemonObservation, now time.Time) (string, bool) { | |
| 220 | + | if daemon.Availability == domain.RemoteAvailabilityNoDaemon { | |
| 221 | + | return "no vev daemon — Enter in the picker to create a session", false | |
| 222 | + | } | |
| 223 | + | if daemon.Availability == domain.RemoteAvailabilityUnknown { | |
| 224 | + | return "not yet observed", false | |
| 225 | + | } | |
| 226 | + | if daemon.Availability == domain.RemoteAvailabilityIncompatible { | |
| 227 | + | return "incompatible (identity changed? use vev host rm/add)", false | |
| 228 | + | } | |
| 229 | + | if daemon.Availability != domain.RemoteAvailabilityReachable { | |
| 230 | + | return "down (" + daemon.Availability.String() + ")", false | |
| 231 | + | } | |
| 232 | + | if daemon.LastSuccess.IsZero() || now.Sub(daemon.LastSuccess) > 15*time.Second { | |
| 233 | + | if daemon.LastSuccess.IsZero() { | |
| 234 | + | return "stale (last observation unknown)", false | |
| 235 | + | } | |
| 236 | + | return fmt.Sprintf("stale (%s since last observation)", now.Sub(daemon.LastSuccess).Round(time.Second)), false | |
| 237 | + | } | |
| 238 | + | return "reachable", true | |
| 239 | + | } | |
| 240 | + | ||
| 156 | 241 | func catalogSessionsAsInfo(host string, sessions []catalogue.RemoteCatalogSession) []protocol.SessionInfo { | |
| 157 | 242 | out := make([]protocol.SessionInfo, 0, len(sessions)) | |
| 158 | 243 | for _, session := range sessions { | |
| @@ -2,12 +2,16 @@ package app | |||||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | "context" | 4 | "context" | ||
| 5 | + | "errors" | |||
| 5 | "fmt" | 6 | "fmt" | ||
| 6 | "io" | 7 | "io" | ||
| 7 | "os" | 8 | "os" | ||
| 8 | "sort" | 9 | "sort" | ||
| 9 | "text/tabwriter" | 10 | "text/tabwriter" | ||
| 11 | + | "time" | |||
| 10 | 12 | ||||
| 13 | + | "github.com/bnema/vev/internal/adapters/brokerconfig" | |||
| 14 | + | "github.com/bnema/vev/internal/adapters/brokerstore" | |||
| 11 | "github.com/bnema/vev/internal/domain" | 15 | "github.com/bnema/vev/internal/domain" | ||
| 12 | "github.com/bnema/vev/internal/ports" | 16 | "github.com/bnema/vev/internal/ports" | ||
| 13 | "github.com/bnema/vev/internal/protocol" | 17 | "github.com/bnema/vev/internal/protocol" | ||
| @@ -21,18 +25,22 @@ const ( | |||||
| 21 | ) | 25 | ) | ||
| 22 | 26 | ||||
| 23 | type remoteHostDeps struct { | 27 | type remoteHostDeps struct { | ||
| 24 | - | connect func(context.Context) (ports.BrokerService, error) | 28 | + | connect func(context.Context) (ports.BrokerService, error) |
| 25 | - | stdout io.Writer | 29 | + | connectExisting func(context.Context) (ports.BrokerService, error) |
| 30 | + | stdout io.Writer | |||
| 26 | } | 31 | } | ||
| 27 | 32 | ||||
| 28 | func defaultRemoteHostDeps() remoteHostDeps { | 33 | func defaultRemoteHostDeps() remoteHostDeps { | ||
| 29 | - | return remoteHostDeps{connect: connectProductionBroker, stdout: os.Stdout} | 34 | + | return remoteHostDeps{connect: connectProductionBroker, connectExisting: connectExistingBroker, stdout: os.Stdout} |
| 30 | } | 35 | } | ||
| 31 | 36 | ||||
| 32 | func (d remoteHostDeps) withDefaults() remoteHostDeps { | 37 | func (d remoteHostDeps) withDefaults() remoteHostDeps { | ||
| 33 | if d.connect == nil { | 38 | if d.connect == nil { | ||
| 34 | d.connect = connectProductionBroker | 39 | d.connect = connectProductionBroker | ||
| 35 | } | 40 | } | ||
| 41 | + | if d.connectExisting == nil { | |||
| 42 | + | d.connectExisting = connectExistingBroker | |||
| 43 | + | } | |||
| 36 | if d.stdout == nil { | 44 | if d.stdout == nil { | ||
| 37 | d.stdout = os.Stdout | 45 | d.stdout = os.Stdout | ||
| 38 | } | 46 | } | ||
| @@ -41,8 +49,15 @@ func (d remoteHostDeps) withDefaults() remoteHostDeps { | |||||
| 41 | 49 | ||||
| 42 | func runHostCommand(ctx context.Context, cmd command, deps remoteHostDeps) error { | 50 | func runHostCommand(ctx context.Context, cmd command, deps remoteHostDeps) error { | ||
| 43 | deps = deps.withDefaults() | 51 | deps = deps.withDefaults() | ||
| 44 | - | service, err := deps.connect(ctx) | 52 | + | connect := deps.connect |
| 53 | + | if cmd.hostAction == hostActionList { | |||
| 54 | + | connect = deps.connectExisting | |||
| 55 | + | } | |||
| 56 | + | service, err := connect(ctx) | |||
| 45 | if err != nil { | 57 | if err != nil { | ||
| 58 | + | if cmd.hostAction == hostActionList && errors.Is(err, errBrokerAbsent) { | |||
| 59 | + | return listDisconnectedHosts(deps.stdout) | |||
| 60 | + | } | |||
| 46 | return err | 61 | return err | ||
| 47 | } | 62 | } | ||
| 48 | defer func() { _ = service.Close() }() | 63 | defer func() { _ = service.Close() }() | ||
| @@ -86,6 +101,42 @@ func runHostCommand(ctx context.Context, cmd command, deps remoteHostDeps) error | |||||
| 86 | } | 101 | } | ||
| 87 | } | 102 | } | ||
| 88 | 103 | ||||
| 104 | + | // listDisconnectedHosts reads configured endpoints without dialing any remote. | |||
| 105 | + | func listDisconnectedHosts(w io.Writer) error { | |||
| 106 | + | state := productionBrokerLayout().State | |||
| 107 | + | hosts, err := brokerstore.ReadHosts(state) | |||
| 108 | + | var endpoints []string | |||
| 109 | + | if errors.Is(err, os.ErrNotExist) { | |||
| 110 | + | // Before the first broker run, the config is the only membership. | |||
| 111 | + | config, configErr := brokerconfig.LoadPath(productionBrokerLayout(), productionBrokerConfigPath()) | |||
| 112 | + | if errors.Is(configErr, os.ErrNotExist) { | |||
| 113 | + | _, err = fmt.Fprintln(w, "no hosts") | |||
| 114 | + | return err | |||
| 115 | + | } | |||
| 116 | + | if configErr != nil { | |||
| 117 | + | return configErr | |||
| 118 | + | } | |||
| 119 | + | endpoints = config.Endpoints() | |||
| 120 | + | } else if err != nil { | |||
| 121 | + | return err | |||
| 122 | + | } else { | |||
| 123 | + | for _, host := range hosts.Hosts { | |||
| 124 | + | endpoints = append(endpoints, host.Registration.Endpoint) | |||
| 125 | + | } | |||
| 126 | + | sort.Strings(endpoints) | |||
| 127 | + | } | |||
| 128 | + | if len(endpoints) == 0 { | |||
| 129 | + | _, err = fmt.Fprintln(w, "no hosts") | |||
| 130 | + | return err | |||
| 131 | + | } | |||
| 132 | + | for _, endpoint := range endpoints { | |||
| 133 | + | if _, err := fmt.Fprintf(w, "%s: not connected (no broker running)\n", endpoint); err != nil { | |||
| 134 | + | return err | |||
| 135 | + | } | |||
| 136 | + | } | |||
| 137 | + | return nil | |||
| 138 | + | } | |||
| 139 | + | ||||
| 89 | func printBrokerHosts(w io.Writer, snapshot ports.BrokerSnapshot) error { | 140 | func printBrokerHosts(w io.Writer, snapshot ports.BrokerSnapshot) error { | ||
| 90 | if err := snapshot.Validate(); err != nil { | 141 | if err := snapshot.Validate(); err != nil { | ||
| 91 | return fmt.Errorf("vev: reading broker catalogue: %w", err) | 142 | return fmt.Errorf("vev: reading broker catalogue: %w", err) | ||
| @@ -104,10 +155,10 @@ func printBrokerHosts(w io.Writer, snapshot ports.BrokerSnapshot) error { | |||||
| 104 | tw := tabwriter.NewWriter(w, 0, 4, 2, ' ', 0) | 155 | tw := tabwriter.NewWriter(w, 0, 4, 2, ' ', 0) | ||
| 105 | _, _ = fmt.Fprintln(tw, "TARGET\tSOURCE\tSTATUS") | 156 | _, _ = fmt.Fprintln(tw, "TARGET\tSOURCE\tSTATUS") | ||
| 106 | for _, endpoint := range endpoints { | 157 | for _, endpoint := range endpoints { | ||
| 107 | - | status := "configured" | 158 | + | status := "not connected" |
| 108 | for _, daemon := range snapshot.Daemons { | 159 | for _, daemon := range snapshot.Daemons { | ||
| 109 | - | if !daemon.Local && daemon.Endpoint == endpoint && daemon.Availability == domain.RemoteAvailabilityNoDaemon { | 160 | + | if !daemon.Local && daemon.Endpoint == endpoint { |
| 110 | - | status = "no daemon" | 161 | + | status, _ = observedHostStatus(daemon, time.Now()) |
| 111 | break | 162 | break | ||
| 112 | } | 163 | } | ||
| 113 | } | 164 | } | ||
| @@ -117,6 +168,10 @@ func printBrokerHosts(w io.Writer, snapshot ports.BrokerSnapshot) error { | |||||
| 117 | } | 168 | } | ||
| 118 | 169 | ||||
| 119 | func runBrokerSnapshotList(cmd command, snapshot ports.BrokerSnapshot, stdout io.Writer) error { | 170 | func runBrokerSnapshotList(cmd command, snapshot ports.BrokerSnapshot, stdout io.Writer) error { | ||
| 171 | + | return renderBrokerSnapshotList(cmd, snapshot, stdout, time.Now()) | |||
| 172 | + | } | |||
| 173 | + | ||||
| 174 | + | func renderBrokerSnapshotList(cmd command, snapshot ports.BrokerSnapshot, stdout io.Writer, now time.Time) error { | |||
| 120 | if err := snapshot.Validate(); err != nil { | 175 | if err := snapshot.Validate(); err != nil { | ||
| 121 | return fmt.Errorf("vev: reading broker catalogue: %w", err) | 176 | return fmt.Errorf("vev: reading broker catalogue: %w", err) | ||
| 122 | } | 177 | } | ||
| @@ -128,8 +183,15 @@ func runBrokerSnapshotList(cmd command, snapshot ports.BrokerSnapshot, stdout io | |||||
| 128 | continue | 183 | continue | ||
| 129 | } | 184 | } | ||
| 130 | found = true | 185 | found = true | ||
| 131 | - | if !daemon.Local && daemon.Availability == domain.RemoteAvailabilityNoDaemon { | 186 | + | status, showSessions := observedHostStatus(daemon, now) |
| 132 | - | statuses = append(statuses, fmt.Sprintf("%s: no vev daemon — Enter in the picker to create a session", daemon.Endpoint)) | 187 | + | if daemon.Local { |
| 188 | + | showSessions = daemon.Availability == domain.RemoteAvailabilityReachable | |||
| 189 | + | } | |||
| 190 | + | if !daemon.Local && status != "reachable" { | |||
| 191 | + | statuses = append(statuses, fmt.Sprintf("%s: %s", daemon.Endpoint, status)) | |||
| 192 | + | } | |||
| 193 | + | if !showSessions { | |||
| 194 | + | continue | |||
| 133 | } | 195 | } | ||
| 134 | for i, session := range catalogSessionsAsInfo(daemon.Endpoint, daemon.Sessions) { | 196 | for i, session := range catalogSessionsAsInfo(daemon.Endpoint, daemon.Sessions) { | ||
| 135 | if daemon.Local { | 197 | if daemon.Local { | ||
| @@ -153,6 +215,29 @@ func runBrokerSnapshotList(cmd command, snapshot ports.BrokerSnapshot, stdout io | |||||
| 153 | return nil | 215 | return nil | ||
| 154 | } | 216 | } | ||
| 155 | 217 | ||||
| 218 | + | // observedHostStatus never treats cached inventory as proof of current reachability. | |||
| 219 | + | func observedHostStatus(daemon ports.BrokerDaemonObservation, now time.Time) (string, bool) { | |||
| 220 | + | if daemon.Availability == domain.RemoteAvailabilityNoDaemon { | |||
| 221 | + | return "no vev daemon — Enter in the picker to create a session", false | |||
| 222 | + | } | |||
| 223 | + | if daemon.Availability == domain.RemoteAvailabilityUnknown { | |||
| 224 | + | return "not yet observed", false | |||
| 225 | + | } | |||
| 226 | + | if daemon.Availability == domain.RemoteAvailabilityIncompatible { | |||
| 227 | + | return "incompatible (identity changed? use vev host rm/add)", false | |||
| 228 | + | } | |||
| 229 | + | if daemon.Availability != domain.RemoteAvailabilityReachable { | |||
| 230 | + | return "down (" + daemon.Availability.String() + ")", false | |||
| 231 | + | } | |||
| 232 | + | if daemon.LastSuccess.IsZero() || now.Sub(daemon.LastSuccess) > 15*time.Second { | |||
| 233 | + | if daemon.LastSuccess.IsZero() { | |||
| 234 | + | return "stale (last observation unknown)", false | |||
| 235 | + | } | |||
| 236 | + | return fmt.Sprintf("stale (%s since last observation)", now.Sub(daemon.LastSuccess).Round(time.Second)), false | |||
| 237 | + | } | |||
| 238 | + | return "reachable", true | |||
| 239 | + | } | |||
| 240 | + | ||||
| 156 | func catalogSessionsAsInfo(host string, sessions []catalogue.RemoteCatalogSession) []protocol.SessionInfo { | 241 | func catalogSessionsAsInfo(host string, sessions []catalogue.RemoteCatalogSession) []protocol.SessionInfo { | ||
| 157 | out := make([]protocol.SessionInfo, 0, len(sessions)) | 242 | out := make([]protocol.SessionInfo, 0, len(sessions)) | ||
| 158 | for _, session := range sessions { | 243 | for _, session := range sessions { | ||
M internal/app/run.go +48 -3
| @@ -1303,10 +1303,12 @@ func copyTransport(ctx context.Context, src, dst wire.Transport) error { | |||
| 1303 | 1303 | } | |
| 1304 | 1304 | } | |
| 1305 | 1305 | ||
| 1306 | - | // runList obtains local live data or a multi-host catalogue snapshot through | |
| 1307 | - | // the per-user broker. It never dials a daemon directly. | |
| 1306 | + | // runList reads existing state without starting either background process. | |
| 1308 | 1307 | func runList(ctx context.Context, cmd command) error { | |
| 1309 | - | service, err := connectBroker(ctx) | |
| 1308 | + | service, err := connectListBroker(ctx) | |
| 1309 | + | if errors.Is(err, errBrokerAbsent) { | |
| 1310 | + | return runListWithoutBroker(ctx, cmd, os.Stdout) | |
| 1311 | + | } | |
| 1310 | 1312 | if err != nil { | |
| 1311 | 1313 | return unreachableBrokerError(err) | |
| 1312 | 1314 | } | |
| @@ -1320,12 +1322,55 @@ func runList(ctx context.Context, cmd command) error { | |||
| 1320 | 1322 | } | |
| 1321 | 1323 | sessions, err := operations.List(ctx, localBrokerOperationRoute(service.Snapshot())) | |
| 1322 | 1324 | if err != nil { | |
| 1325 | + | var brokerErr ports.BrokerError | |
| 1326 | + | if errors.As(err, &brokerErr) && brokerErr.Code == ports.BrokerErrorNoDaemon { | |
| 1327 | + | printSessions(os.Stdout, nil) | |
| 1328 | + | return nil | |
| 1329 | + | } | |
| 1323 | 1330 | return err | |
| 1324 | 1331 | } | |
| 1325 | 1332 | printSessions(os.Stdout, sessions) | |
| 1326 | 1333 | return nil | |
| 1327 | 1334 | } | |
| 1328 | 1335 | ||
| 1336 | + | var connectListBroker = connectExistingBroker | |
| 1337 | + | var dialListDaemon = ipc.DialContext | |
| 1338 | + | ||
| 1339 | + | func runListWithoutBroker(ctx context.Context, cmd command, out io.Writer) error { | |
| 1340 | + | if cmd.listHost != "" { | |
| 1341 | + | _, err := fmt.Fprintf(out, "%s: not connected (no broker running)\n", cmd.listHost) | |
| 1342 | + | return err | |
| 1343 | + | } | |
| 1344 | + | transport, err := dialListDaemon(ctx, ipc.SocketDir()) | |
| 1345 | + | if err != nil { | |
| 1346 | + | if !backendAbsent(err) { | |
| 1347 | + | return fmt.Errorf("vev: list local sessions: %w", err) | |
| 1348 | + | } | |
| 1349 | + | } else { | |
| 1350 | + | connection := sessionwire.NewClientConnection(transport) | |
| 1351 | + | defer func() { _ = connection.Close() }() | |
| 1352 | + | if err := connection.SendClient(protocol.List{}); err != nil { | |
| 1353 | + | return fmt.Errorf("vev: request local sessions: %w", err) | |
| 1354 | + | } | |
| 1355 | + | reply, err := connection.ReceiveServer() | |
| 1356 | + | if err != nil { | |
| 1357 | + | return fmt.Errorf("vev: read local sessions: %w", err) | |
| 1358 | + | } | |
| 1359 | + | sessions, ok := reply.(protocol.Sessions) | |
| 1360 | + | if !ok { | |
| 1361 | + | return fmt.Errorf("vev: unexpected local list reply %T", reply) | |
| 1362 | + | } | |
| 1363 | + | printSessions(out, sessions.Sessions) | |
| 1364 | + | } | |
| 1365 | + | if err != nil { | |
| 1366 | + | printSessions(out, nil) | |
| 1367 | + | } | |
| 1368 | + | if cmd.listAll { | |
| 1369 | + | return listDisconnectedHosts(out) | |
| 1370 | + | } | |
| 1371 | + | return nil | |
| 1372 | + | } | |
| 1373 | + | ||
| 1329 | 1374 | // printSessions renders a session table (or a friendly note when empty). | |
| 1330 | 1375 | func printSessions(w io.Writer, sessions []protocol.SessionInfo) { | |
| 1331 | 1376 | if len(sessions) == 0 { | |
| @@ -1303,10 +1303,12 @@ func copyTransport(ctx context.Context, src, dst wire.Transport) error { | |||||
| 1303 | } | 1303 | } | ||
| 1304 | } | 1304 | } | ||
| 1305 | 1305 | ||||
| 1306 | - | // runList obtains local live data or a multi-host catalogue snapshot through | 1306 | + | // runList reads existing state without starting either background process. |
| 1307 | - | // the per-user broker. It never dials a daemon directly. | |||
| 1308 | func runList(ctx context.Context, cmd command) error { | 1307 | func runList(ctx context.Context, cmd command) error { | ||
| 1309 | - | service, err := connectBroker(ctx) | 1308 | + | service, err := connectListBroker(ctx) |
| 1309 | + | if errors.Is(err, errBrokerAbsent) { | |||
| 1310 | + | return runListWithoutBroker(ctx, cmd, os.Stdout) | |||
| 1311 | + | } | |||
| 1310 | if err != nil { | 1312 | if err != nil { | ||
| 1311 | return unreachableBrokerError(err) | 1313 | return unreachableBrokerError(err) | ||
| 1312 | } | 1314 | } | ||
| @@ -1320,12 +1322,55 @@ func runList(ctx context.Context, cmd command) error { | |||||
| 1320 | } | 1322 | } | ||
| 1321 | sessions, err := operations.List(ctx, localBrokerOperationRoute(service.Snapshot())) | 1323 | sessions, err := operations.List(ctx, localBrokerOperationRoute(service.Snapshot())) | ||
| 1322 | if err != nil { | 1324 | if err != nil { | ||
| 1325 | + | var brokerErr ports.BrokerError | |||
| 1326 | + | if errors.As(err, &brokerErr) && brokerErr.Code == ports.BrokerErrorNoDaemon { | |||
| 1327 | + | printSessions(os.Stdout, nil) | |||
| 1328 | + | return nil | |||
| 1329 | + | } | |||
| 1323 | return err | 1330 | return err | ||
| 1324 | } | 1331 | } | ||
| 1325 | printSessions(os.Stdout, sessions) | 1332 | printSessions(os.Stdout, sessions) | ||
| 1326 | return nil | 1333 | return nil | ||
| 1327 | } | 1334 | } | ||
| 1328 | 1335 | ||||
| 1336 | + | var connectListBroker = connectExistingBroker | |||
| 1337 | + | var dialListDaemon = ipc.DialContext | |||
| 1338 | + | ||||
| 1339 | + | func runListWithoutBroker(ctx context.Context, cmd command, out io.Writer) error { | |||
| 1340 | + | if cmd.listHost != "" { | |||
| 1341 | + | _, err := fmt.Fprintf(out, "%s: not connected (no broker running)\n", cmd.listHost) | |||
| 1342 | + | return err | |||
| 1343 | + | } | |||
| 1344 | + | transport, err := dialListDaemon(ctx, ipc.SocketDir()) | |||
| 1345 | + | if err != nil { | |||
| 1346 | + | if !backendAbsent(err) { | |||
| 1347 | + | return fmt.Errorf("vev: list local sessions: %w", err) | |||
| 1348 | + | } | |||
| 1349 | + | } else { | |||
| 1350 | + | connection := sessionwire.NewClientConnection(transport) | |||
| 1351 | + | defer func() { _ = connection.Close() }() | |||
| 1352 | + | if err := connection.SendClient(protocol.List{}); err != nil { | |||
| 1353 | + | return fmt.Errorf("vev: request local sessions: %w", err) | |||
| 1354 | + | } | |||
| 1355 | + | reply, err := connection.ReceiveServer() | |||
| 1356 | + | if err != nil { | |||
| 1357 | + | return fmt.Errorf("vev: read local sessions: %w", err) | |||
| 1358 | + | } | |||
| 1359 | + | sessions, ok := reply.(protocol.Sessions) | |||
| 1360 | + | if !ok { | |||
| 1361 | + | return fmt.Errorf("vev: unexpected local list reply %T", reply) | |||
| 1362 | + | } | |||
| 1363 | + | printSessions(out, sessions.Sessions) | |||
| 1364 | + | } | |||
| 1365 | + | if err != nil { | |||
| 1366 | + | printSessions(out, nil) | |||
| 1367 | + | } | |||
| 1368 | + | if cmd.listAll { | |||
| 1369 | + | return listDisconnectedHosts(out) | |||
| 1370 | + | } | |||
| 1371 | + | return nil | |||
| 1372 | + | } | |||
| 1373 | + | ||||
| 1329 | // printSessions renders a session table (or a friendly note when empty). | 1374 | // printSessions renders a session table (or a friendly note when empty). | ||
| 1330 | func printSessions(w io.Writer, sessions []protocol.SessionInfo) { | 1375 | func printSessions(w io.Writer, sessions []protocol.SessionInfo) { | ||
| 1331 | if len(sessions) == 0 { | 1376 | if len(sessions) == 0 { | ||
M internal/app/run_broker_test.go +5 -2
| @@ -205,11 +205,14 @@ var _ ports.BrokerService = (*seamBrokerService)(nil) | |||
| 205 | 205 | func withSeamBroker(t *testing.T, service ports.BrokerService) { | |
| 206 | 206 | t.Helper() | |
| 207 | 207 | previous := connectBroker | |
| 208 | + | previousList := connectListBroker | |
| 208 | 209 | previousStop := connectDaemonStopBroker | |
| 209 | 210 | connectBroker = func(context.Context) (ports.BrokerService, error) { return service, nil } | |
| 211 | + | connectListBroker = connectBroker | |
| 210 | 212 | connectDaemonStopBroker = connectBroker | |
| 211 | 213 | t.Cleanup(func() { | |
| 212 | 214 | connectBroker = previous | |
| 215 | + | connectListBroker = previousList | |
| 213 | 216 | connectDaemonStopBroker = previousStop | |
| 214 | 217 | }) | |
| 215 | 218 | } | |
| @@ -276,8 +279,8 @@ func TestRunListRendersBrokerSessionsWithoutADaemon(t *testing.T) { | |||
| 276 | 279 | require.Len(t, opens, 1) | |
| 277 | 280 | require.Equal(t, ports.BrokerStreamControl, opens[0].Purpose) | |
| 278 | 281 | require.True(t, opens[0].Local) | |
| 279 | - | require.Equal(t, ports.BrokerDaemonStartIfNeeded, opens[0].StartMode, | |
| 280 | - | "a local listing may start a stopped daemon to read its records") | |
| 282 | + | require.Equal(t, ports.BrokerDaemonExistingOnly, opens[0].StartMode, | |
| 283 | + | "a listing must not start a stopped daemon") | |
| 281 | 284 | } | |
| 282 | 285 | ||
| 283 | 286 | // TestRunKillExplicitOutcomesTable pins the runKill outcome contract: an | |
| @@ -205,11 +205,14 @@ var _ ports.BrokerService = (*seamBrokerService)(nil) | |||||
| 205 | func withSeamBroker(t *testing.T, service ports.BrokerService) { | 205 | func withSeamBroker(t *testing.T, service ports.BrokerService) { | ||
| 206 | t.Helper() | 206 | t.Helper() | ||
| 207 | previous := connectBroker | 207 | previous := connectBroker | ||
| 208 | + | previousList := connectListBroker | |||
| 208 | previousStop := connectDaemonStopBroker | 209 | previousStop := connectDaemonStopBroker | ||
| 209 | connectBroker = func(context.Context) (ports.BrokerService, error) { return service, nil } | 210 | connectBroker = func(context.Context) (ports.BrokerService, error) { return service, nil } | ||
| 211 | + | connectListBroker = connectBroker | |||
| 210 | connectDaemonStopBroker = connectBroker | 212 | connectDaemonStopBroker = connectBroker | ||
| 211 | t.Cleanup(func() { | 213 | t.Cleanup(func() { | ||
| 212 | connectBroker = previous | 214 | connectBroker = previous | ||
| 215 | + | connectListBroker = previousList | |||
| 213 | connectDaemonStopBroker = previousStop | 216 | connectDaemonStopBroker = previousStop | ||
| 214 | }) | 217 | }) | ||
| 215 | } | 218 | } | ||
| @@ -276,8 +279,8 @@ func TestRunListRendersBrokerSessionsWithoutADaemon(t *testing.T) { | |||||
| 276 | require.Len(t, opens, 1) | 279 | require.Len(t, opens, 1) | ||
| 277 | require.Equal(t, ports.BrokerStreamControl, opens[0].Purpose) | 280 | require.Equal(t, ports.BrokerStreamControl, opens[0].Purpose) | ||
| 278 | require.True(t, opens[0].Local) | 281 | require.True(t, opens[0].Local) | ||
| 279 | - | require.Equal(t, ports.BrokerDaemonStartIfNeeded, opens[0].StartMode, | 282 | + | require.Equal(t, ports.BrokerDaemonExistingOnly, opens[0].StartMode, |
| 280 | - | "a local listing may start a stopped daemon to read its records") | 283 | + | "a listing must not start a stopped daemon") |
| 281 | } | 284 | } | ||
| 282 | 285 | ||||
| 283 | // TestRunKillExplicitOutcomesTable pins the runKill outcome contract: an | 286 | // TestRunKillExplicitOutcomesTable pins the runKill outcome contract: an | ||
M internal/usecase/client/broker_operations.go +1 -1
| @@ -123,7 +123,7 @@ func NewBrokerOperations(service ports.BrokerService, clock ports.Clock) (*Broke | |||
| 123 | 123 | // unreachable: a refused or lost listing is an ordinary error. A new explicit | |
| 124 | 124 | // list is always safe, so the implementation never retries on its own. | |
| 125 | 125 | func (o *BrokerOperations) List(ctx context.Context, route BrokerOperationRoute) ([]protocol.SessionInfo, error) { | |
| 126 | - | reply, err := o.exchange(ctx, route, ports.BrokerDaemonStartIfNeeded, protocol.List{}) | |
| 126 | + | reply, err := o.exchange(ctx, route, ports.BrokerDaemonExistingOnly, protocol.List{}) | |
| 127 | 127 | if err != nil { | |
| 128 | 128 | if errors.Is(err, errBrokerOperationLost) { | |
| 129 | 129 | return nil, fmt.Errorf("vev: reading session list: %w", err) | |
| @@ -123,7 +123,7 @@ func NewBrokerOperations(service ports.BrokerService, clock ports.Clock) (*Broke | |||||
| 123 | // unreachable: a refused or lost listing is an ordinary error. A new explicit | 123 | // unreachable: a refused or lost listing is an ordinary error. A new explicit | ||
| 124 | // list is always safe, so the implementation never retries on its own. | 124 | // list is always safe, so the implementation never retries on its own. | ||
| 125 | func (o *BrokerOperations) List(ctx context.Context, route BrokerOperationRoute) ([]protocol.SessionInfo, error) { | 125 | func (o *BrokerOperations) List(ctx context.Context, route BrokerOperationRoute) ([]protocol.SessionInfo, error) { | ||
| 126 | - | reply, err := o.exchange(ctx, route, ports.BrokerDaemonStartIfNeeded, protocol.List{}) | 126 | + | reply, err := o.exchange(ctx, route, ports.BrokerDaemonExistingOnly, protocol.List{}) |
| 127 | if err != nil { | 127 | if err != nil { | ||
| 128 | if errors.Is(err, errBrokerOperationLost) { | 128 | if errors.Is(err, errBrokerOperationLost) { | ||
| 129 | return nil, fmt.Errorf("vev: reading session list: %w", err) | 129 | return nil, fmt.Errorf("vev: reading session list: %w", err) | ||
M internal/usecase/client/broker_operations_test.go +1 -1
| @@ -480,7 +480,7 @@ func TestBrokerOperationsListUsesOneControlStream(t *testing.T) { | |||
| 480 | 480 | require.Empty(t, request.Name) | |
| 481 | 481 | require.Equal(t, protocol.ExactSessionTarget{}, request.Target) | |
| 482 | 482 | require.Empty(t, request.Env) | |
| 483 | - | require.Equal(t, ports.BrokerDaemonStartIfNeeded, request.StartMode) | |
| 483 | + | require.Equal(t, ports.BrokerDaemonExistingOnly, request.StartMode) | |
| 484 | 484 | require.True(t, request.Local) | |
| 485 | 485 | require.Equal(t, service.ConnectionID(), request.Connection) | |
| 486 | 486 | require.NotZero(t, request.Stream) | |
| @@ -480,7 +480,7 @@ func TestBrokerOperationsListUsesOneControlStream(t *testing.T) { | |||||
| 480 | require.Empty(t, request.Name) | 480 | require.Empty(t, request.Name) | ||
| 481 | require.Equal(t, protocol.ExactSessionTarget{}, request.Target) | 481 | require.Equal(t, protocol.ExactSessionTarget{}, request.Target) | ||
| 482 | require.Empty(t, request.Env) | 482 | require.Empty(t, request.Env) | ||
| 483 | - | require.Equal(t, ports.BrokerDaemonStartIfNeeded, request.StartMode) | 483 | + | require.Equal(t, ports.BrokerDaemonExistingOnly, request.StartMode) |
| 484 | require.True(t, request.Local) | 484 | require.True(t, request.Local) | ||
| 485 | require.Equal(t, service.ConnectionID(), request.Connection) | 485 | require.Equal(t, service.ConnectionID(), request.Connection) | ||
| 486 | require.NotZero(t, request.Stream) | 486 | require.NotZero(t, request.Stream) | ||