broker_offline_local_attach_test.go RawHistoryFolder
1//go:build linux
2
3package app
4
5import (
6 "context"
7 "os"
8 "path/filepath"
9 "sync"
10 "testing"
11 "time"
12
13 "github.com/stretchr/testify/require"
14
15 "github.com/bnema/vev/internal/adapters/brokerconfig"
16 "github.com/bnema/vev/internal/adapters/clock"
17 "github.com/bnema/vev/internal/adapters/daemonidentity"
18 "github.com/bnema/vev/internal/adapters/daemonmux"
19 "github.com/bnema/vev/internal/adapters/ipc"
20 "github.com/bnema/vev/internal/adapters/pty"
21 "github.com/bnema/vev/internal/adapters/sessionwire"
22 "github.com/bnema/vev/internal/domain"
23 "github.com/bnema/vev/internal/ports"
24 "github.com/bnema/vev/internal/protocol"
25 "github.com/bnema/vev/internal/protocol/catalogue"
26 "github.com/bnema/vev/internal/usecase/broker"
27 "github.com/bnema/vev/internal/usecase/daemon"
28)
29
30// P7.4b real-composition coverage for the broker-owned local probe.
31//
32// The scripted tests prove the probe's exact stream semantics. These tests run
33// the same probe against a real daemon serving a real private Unix daemonmux
34// carriage, so the authenticated physical preamble, the observation open, the
35// typed remote-catalog command, and the catalogue validation all compose against
36// production daemon code. They prove the observation is real: it creates no
37// session or process and does not change daemon-owned identity; the catalogue it publishes
38// is exactly what makes an exact terminal attach resolvable.
39
40// realLocalDaemonFixture is a real daemon serving one private Unix daemonmux
41// carriage, plus the production daemon state directory a probe must never mutate.
42type realLocalDaemonFixture struct {
43 binding brokerconfig.LocalBinding
44 route string
45 stateDir string
46 connector *daemonmux.EndpointConnector
47 // nextStream allocates the strictly increasing logical stream identity each
48 // helper attachment names, as the daemonmux engine requires within one
49 // physical connection.
50 nextStream ports.BrokerStreamID
51}
52
53// startRealLocalDaemon starts one real daemon over a private local daemonmux
54// carriage and returns the fixture the probe dials.
55func startRealLocalDaemon(t *testing.T) *realLocalDaemonFixture {
56 t.Helper()
57 layout := emptyProductionBrokerLayout(t, "")
58 stateDir := filepath.Dir(layout.Root)
59 identity, err := daemonidentity.LoadOrCreate(stateDir)
60 require.NoError(t, err)
61 config, err := brokerconfig.LoadProduction(layout, productionBrokerConfigPath(), identity, localDaemonPolicy(), daemonmux.SocketPath(ipc.SocketDir()))
62 require.NoError(t, err)
63 binding, ok := config.LocalBinding()
64 require.True(t, ok)
65 route := binding.Route.Path()
66 require.NoError(t, os.MkdirAll(filepath.Dir(route), 0o700))
67
68 listener, err := ipc.ListenMux(route)
69 require.NoError(t, err)
70 aggregate := daemonmux.NewAggregateListener()
71 serverBinding := testServerBinding(t, binding)
72 supervisor, err := daemonmux.NewServerSupervisor(aggregate, serverBinding, daemonmux.DefaultMuxCeilings(), 0)
73 require.NoError(t, err)
74
75 ctx, cancel := context.WithCancel(context.Background())
76 d := daemon.New(pty.NewFactory(), clock.New(), discardLog(),
77 daemon.WithShell("/bin/sh", []string{"-c", "trap : TERM; while :; do read line || exit; done"}))
78 go func() { _ = d.Serve(ctx, aggregate) }()
79 go func() {
80 for {
81 raw, acceptErr := listener.Accept()
82 if acceptErr != nil {
83 return
84 }
85 go func() { _ = supervisor.Adopt(ctx, raw) }()
86 }
87 }()
88 t.Cleanup(func() {
89 cancel()
90 _ = listener.Close()
91 _ = supervisor.Close()
92 _ = aggregate.Close()
93 })
94
95 dial := func(ctx context.Context, _ ports.BrokerDialTarget) (daemonmux.RawFramedTransport, error) {
96 return ipc.DialMuxContext(ctx, route)
97 }
98 connector, err := daemonmux.NewEndpointConnector(daemonmux.RawCarrierDialer(dial), daemonmux.DefaultMuxCeilings())
99 require.NoError(t, err)
100 return &realLocalDaemonFixture{binding: binding, route: route, stateDir: stateDir, connector: connector}
101}
102
103// probe returns one probe over the fixture's real carriage.
104func (f *realLocalDaemonFixture) probe(t *testing.T) *broker.LocalDaemonProbe {
105 t.Helper()
106 dial := func(ctx context.Context, target ports.BrokerDialTarget) (daemonmux.RawFramedTransport, error) {
107 require.Equal(t, f.binding.Route.Address(), target.Address, "the probe dials only its provisioned local address")
108 return ipc.DialMuxContext(ctx, f.route)
109 }
110 probe, err := newLocalRouteProbe(f.binding, brokerLocalTestEpoch, dial, daemonmux.DefaultMuxCeilings())
111 require.NoError(t, err)
112 return probe
113}
114
115// openStream opens one logical stream over the fixture's real carriage under
116// the fixture's authenticated local endpoint.
117func (f *realLocalDaemonFixture) openStream(t *testing.T, request ports.BrokerOpenStreamRequest) ports.BrokerLogicalConnection {
118 t.Helper()
119 ctx, cancel := context.WithTimeout(context.Background(), brokerTestWait)
120 t.Cleanup(cancel)
121 physical, err := f.connector.Connect(ctx, ports.BrokerDialTarget{
122 Fence: ports.BrokerEndpointFence{Local: true}, Policy: f.binding.Policy, Address: f.binding.Route.Address(),
123 StartMode: ports.BrokerDaemonExistingOnly,
124 ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: f.binding.Identity, Bound: true},
125 })
126 require.NoError(t, err)
127 t.Cleanup(func() { _ = physical.Close() })
128
129 request.Epoch = brokerLocalTestEpoch
130 request.Local = true
131 request.Policy = f.binding.Policy
132 request.Connection = ports.BrokerConnectionID{4, 4}
133 if request.Stream == 0 {
134 f.nextStream++
135 request.Stream = f.nextStream
136 }
137 stream, err := physical.OpenStream(ctx, request)
138 require.NoError(t, err)
139 typed := brokerLogicalView{ClientConnection: sessionwire.BrokerCodec{}.Client(stream), BrokerEnvelopeStream: stream}
140 t.Cleanup(func() { _ = typed.Close() })
141 return typed
142}
143
144// brokerLogicalView is the typed client view of one raw logical stream.
145type brokerLogicalView struct {
146 ports.ClientConnection
147 ports.BrokerEnvelopeStream
148}
149
150func (v brokerLogicalView) Close() error { return v.ClientConnection.Close() }
151
152// awaitWelcome receives server messages until the typed Welcome arrives.
153func awaitWelcome(t *testing.T, stream ports.BrokerLogicalConnection) protocol.Welcome {
154 t.Helper()
155 type outcome struct {
156 message protocol.ServerMessage
157 err error
158 }
159 results := make(chan outcome, 1)
160 go func() {
161 for {
162 message, err := stream.ReceiveServer()
163 if err != nil {
164 results <- outcome{err: err}
165 return
166 }
167 if welcome, ok := message.(protocol.Welcome); ok {
168 results <- outcome{message: welcome}
169 return
170 }
171 }
172 }()
173 select {
174 case result := <-results:
175 require.NoError(t, result.err, "waiting for Welcome")
176 return result.message.(protocol.Welcome)
177 case <-time.After(brokerTestWait):
178 t.Fatal("timed out waiting for Welcome")
179 return protocol.Welcome{}
180 }
181}
182
183// TestBrokerLocalProbeRealCatalogueEnablesExactAttach proves the broker-owned
184// local probe reads a real daemon's catalogue and that the catalogue it
185// publishes is exactly what makes an exact terminal attach resolvable.
186func TestBrokerLocalProbeRealCatalogueEnablesExactAttach(t *testing.T) {
187 fixture := startRealLocalDaemon(t)
188 probe := fixture.probe(t)
189
190 // An observation of a daemon that owns no session reports reachable, known,
191 // and empty inventory: observing never creates a session.
192 empty, err := probe.ProbeLocal(context.Background())
193 require.NoError(t, err)
194 require.Equal(t, domain.RemoteAvailabilityReachable, empty.Availability)
195 require.True(t, empty.InventoryKnown)
196 require.Empty(t, empty.Sessions, "an observation must never create a session")
197 require.Equal(t, fixture.binding.Identity, empty.Identity)
198 require.NotZero(t, empty.ProtocolVersion)
199
200 // One real session is created through one real attachment stream.
201 attachment := fixture.openStream(t, ports.BrokerOpenStreamRequest{
202 Purpose: ports.BrokerStreamAttachment,
203 Admission: ports.BrokerAdmissionCreateNamed,
204 Name: "probe-target",
205 StartMode: ports.BrokerDaemonStartIfNeeded,
206 })
207 require.NoError(t, attachment.SendClient(protocol.Hello{
208 Version: protocol.Version, Intent: protocol.IntentNew, Name: "probe-target",
209 Size: domain.Size{Cols: 80, Rows: 24}, EnvironmentPolicy: protocol.EnvironmentPolicyClientOwned,
210 }))
211 welcome := awaitWelcome(t, attachment)
212 require.NotNil(t, welcome.CommittedIdentity)
213 target := welcome.CommittedIdentity.Target
214
215 // The observation reports the real daemon's catalogue: the exact session
216 // lifecycle the daemon committed, so the observation is the authority an
217 // exact attach resolves against.
218 observed, err := probe.ProbeLocal(context.Background())
219 require.NoError(t, err)
220 require.Equal(t, domain.RemoteAvailabilityReachable, observed.Availability)
221 require.True(t, observed.InventoryKnown)
222 require.Equal(t, fixture.binding.Identity, observed.Identity)
223 require.Len(t, observed.Sessions, 1)
224 require.Equal(t, "probe-target", observed.Sessions[0].Name)
225 require.Equal(t, target.LifecycleID, observed.Sessions[0].LifecycleID)
226 require.Equal(t, catalogue.RemoteCatalogSessionUp, observed.Sessions[0].State)
227
228 // The published catalogue resolves the exact attach target the client
229 // composition uses, and that exact target attaches successfully.
230 resolved, ok := brokerObservationExactTarget(observed, "probe-target")
231 require.True(t, ok)
232 require.Equal(t, target, resolved)
233
234 exact := fixture.openStream(t, ports.BrokerOpenStreamRequest{
235 Purpose: ports.BrokerStreamAttachment,
236 Admission: ports.BrokerAdmissionExact,
237 Target: resolved,
238 StartMode: ports.BrokerDaemonStartIfNeeded,
239 })
240 exactTarget := resolved
241 require.NoError(t, exact.SendClient(protocol.Hello{
242 Version: protocol.Version, Intent: protocol.IntentAttach, Name: resolved.SessionName, ExactTarget: &exactTarget,
243 Size: domain.Size{Cols: 80, Rows: 24}, EnvironmentPolicy: protocol.EnvironmentPolicyClientOwned,
244 }))
245 exactWelcome := awaitWelcome(t, exact)
246 require.NotNil(t, exactWelcome.CommittedIdentity)
247 require.Equal(t, resolved, exactWelcome.CommittedIdentity.Target)
248
249 // The daemon-owned identity persists unchanged after probing.
250 _, statErr := os.Lstat(daemonidentity.Path(fixture.stateDir))
251 require.NoError(t, statErr, "the daemon-owned identity remains after observation")
252}
253
254// TestBrokerLocalProbeAbsentDaemonStaysUnavailable proves that without daemon
255// identity, production observation stays unknown and never dials or starts a daemon.
256func TestBrokerLocalProbeAbsentDaemonStaysUnavailable(t *testing.T) {
257 layout := emptyProductionBrokerLayout(t, "")
258 stateDir := filepath.Dir(layout.Root)
259 config, err := brokerconfig.LoadProduction(layout, productionBrokerConfigPath(), "", localDaemonPolicy(), daemonmux.SocketPath(ipc.SocketDir()))
260 require.NoError(t, err)
261 binding, ok := config.LocalBinding()
262 require.True(t, ok)
263 route := binding.Route.Path()
264
265 var dialed []string
266 var mu sync.Mutex
267 probe, err := newLocalRouteProbe(binding, brokerLocalTestEpoch, func(ctx context.Context, target ports.BrokerDialTarget) (daemonmux.RawFramedTransport, error) {
268 mu.Lock()
269 dialed = append(dialed, target.Address)
270 mu.Unlock()
271 return ipc.DialMuxContext(ctx, binding.Route.Path())
272 }, daemonmux.DefaultMuxCeilings())
273 require.NoError(t, err)
274
275 ctx, cancel := context.WithTimeout(context.Background(), brokerTestWait)
276 defer cancel()
277 observation, err := probe.ProbeLocal(ctx)
278 require.NoError(t, err)
279 require.Equal(t, domain.RemoteAvailabilityUnknown, observation.Availability)
280 require.Zero(t, observation.Identity)
281 require.False(t, observation.InventoryKnown)
282 require.Empty(t, observation.Sessions)
283
284 mu.Lock()
285 require.Empty(t, dialed, "without daemon identity the probe must not dial")
286 mu.Unlock()
287 // Nothing was created: no socket, no identity, no production path entry.
288 _, statErr := os.Lstat(route)
289 require.ErrorIs(t, statErr, os.ErrNotExist, "a probe must never create a carriage")
290 _, identityErr := os.Lstat(daemonidentity.Path(stateDir))
291 require.ErrorIs(t, identityErr, os.ErrNotExist, "a probe must never create daemon identity")
292
293}