fix(client): report the terminal cause and name local refusals
Three defects the P5.4a review recorded but left deferred, all on the path to activating the autonomous client. - A terminal read failure could be reported as an orderly EOF. The reader enqueues its final result for the consumer, but the pump's exit watcher can win that race and used to publish EOF unconditionally, so the supervisor returned a clean status for a terminal that failed. The pump now records the cause the reader ended with and the watcher publishes that. - A selection refused locally was reported as a destination failure, naming a destination that was never dialed. Local refusals now carry their own notice, and the classification order keeps unknown outcomes distinct. - The picker input contract claimed declined reads fall through to ordinary supervisor handling. There is no such handling: a declined read is dropped. The contract now says so. Each fix is mutation-verified: publishing EOF again fails the cause test, and classifying local refusals as destination failures fails the notice test with the wrong kind.
- commit
- 77377555086d4dc839d08a70b24f1fac2271cf55
- author
- brice · · 2026-09-19 13:41 UTC
- parent
- fc4cddc
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
7 files · +138 -16
M internal/usecase/client/client.go +39 -9
| @@ -3627,15 +3627,20 @@ type terminalInputPump struct { | |||
| 3627 | 3627 | activation uint64 | |
| 3628 | 3628 | closed bool | |
| 3629 | 3629 | active bool | |
| 3630 | - | readyMu sync.Mutex | |
| 3631 | - | ready map[uint64]chan struct{} | |
| 3632 | - | claimChanged chan struct{} | |
| 3633 | - | space chan struct{} | |
| 3634 | - | state chan struct{} | |
| 3635 | - | exited chan struct{} | |
| 3636 | - | afterRevoke func() // test synchronization hook | |
| 3637 | - | startMu sync.Once | |
| 3638 | - | stopMu sync.Once | |
| 3630 | + | // readCause stores the cause the terminal reader ended with. The reader | |
| 3631 | + | // enqueues its final result for a consumer to report, but the exit watcher | |
| 3632 | + | // can win that race, so the cause is recorded here as well: a read failure | |
| 3633 | + | // must never be published as an orderly EOF. | |
| 3634 | + | readCause error | |
| 3635 | + | readyMu sync.Mutex | |
| 3636 | + | ready map[uint64]chan struct{} | |
| 3637 | + | claimChanged chan struct{} | |
| 3638 | + | space chan struct{} | |
| 3639 | + | state chan struct{} | |
| 3640 | + | exited chan struct{} | |
| 3641 | + | afterRevoke func() // test synchronization hook | |
| 3642 | + | startMu sync.Once | |
| 3643 | + | stopMu sync.Once | |
| 3639 | 3644 | } | |
| 3640 | 3645 | ||
| 3641 | 3646 | func newTerminalInputPump(in io.Reader) *terminalInputPump { | |
| @@ -3914,6 +3919,7 @@ func (p *terminalInputPump) start() { | |||
| 3914 | 3919 | continue | |
| 3915 | 3920 | } | |
| 3916 | 3921 | if err != nil { | |
| 3922 | + | p.setReadCause(err) | |
| 3917 | 3923 | p.finish() | |
| 3918 | 3924 | return | |
| 3919 | 3925 | } | |
| @@ -3931,6 +3937,30 @@ func (p *terminalInputPump) hasExited() bool { | |||
| 3931 | 3937 | } | |
| 3932 | 3938 | } | |
| 3933 | 3939 | ||
| 3940 | + | // setReadCause records the cause the terminal reader ended with. An orderly EOF | |
| 3941 | + | // is the absence of a failure and stays zero, so cause reports io.EOF for it. | |
| 3942 | + | func (p *terminalInputPump) setReadCause(err error) { | |
| 3943 | + | if err == nil || errors.Is(err, io.EOF) { | |
| 3944 | + | return | |
| 3945 | + | } | |
| 3946 | + | p.mu.Lock() | |
| 3947 | + | if p.readCause == nil { | |
| 3948 | + | p.readCause = err | |
| 3949 | + | } | |
| 3950 | + | p.mu.Unlock() | |
| 3951 | + | } | |
| 3952 | + | ||
| 3953 | + | // cause reports why the terminal read ended: the read failure when there was | |
| 3954 | + | // one, otherwise an orderly EOF. | |
| 3955 | + | func (p *terminalInputPump) cause() error { | |
| 3956 | + | p.mu.Lock() | |
| 3957 | + | defer p.mu.Unlock() | |
| 3958 | + | if p.readCause != nil { | |
| 3959 | + | return p.readCause | |
| 3960 | + | } | |
| 3961 | + | return io.EOF | |
| 3962 | + | } | |
| 3963 | + | ||
| 3934 | 3964 | func (p *terminalInputPump) isClosed() bool { | |
| 3935 | 3965 | p.mu.Lock() | |
| 3936 | 3966 | defer p.mu.Unlock() | |
| @@ -3627,15 +3627,20 @@ type terminalInputPump struct { | |||||
| 3627 | activation uint64 | 3627 | activation uint64 | ||
| 3628 | closed bool | 3628 | closed bool | ||
| 3629 | active bool | 3629 | active bool | ||
| 3630 | - | readyMu sync.Mutex | 3630 | + | // readCause stores the cause the terminal reader ended with. The reader |
| 3631 | - | ready map[uint64]chan struct{} | 3631 | + | // enqueues its final result for a consumer to report, but the exit watcher |
| 3632 | - | claimChanged chan struct{} | 3632 | + | // can win that race, so the cause is recorded here as well: a read failure |
| 3633 | - | space chan struct{} | 3633 | + | // must never be published as an orderly EOF. |
| 3634 | - | state chan struct{} | 3634 | + | readCause error |
| 3635 | - | exited chan struct{} | 3635 | + | readyMu sync.Mutex |
| 3636 | - | afterRevoke func() // test synchronization hook | 3636 | + | ready map[uint64]chan struct{} |
| 3637 | - | startMu sync.Once | 3637 | + | claimChanged chan struct{} |
| 3638 | - | stopMu sync.Once | 3638 | + | space chan struct{} |
| 3639 | + | state chan struct{} | |||
| 3640 | + | exited chan struct{} | |||
| 3641 | + | afterRevoke func() // test synchronization hook | |||
| 3642 | + | startMu sync.Once | |||
| 3643 | + | stopMu sync.Once | |||
| 3639 | } | 3644 | } | ||
| 3640 | 3645 | ||||
| 3641 | func newTerminalInputPump(in io.Reader) *terminalInputPump { | 3646 | func newTerminalInputPump(in io.Reader) *terminalInputPump { | ||
| @@ -3914,6 +3919,7 @@ func (p *terminalInputPump) start() { | |||||
| 3914 | continue | 3919 | continue | ||
| 3915 | } | 3920 | } | ||
| 3916 | if err != nil { | 3921 | if err != nil { | ||
| 3922 | + | p.setReadCause(err) | |||
| 3917 | p.finish() | 3923 | p.finish() | ||
| 3918 | return | 3924 | return | ||
| 3919 | } | 3925 | } | ||
| @@ -3931,6 +3937,30 @@ func (p *terminalInputPump) hasExited() bool { | |||||
| 3931 | } | 3937 | } | ||
| 3932 | } | 3938 | } | ||
| 3933 | 3939 | ||||
| 3940 | + | // setReadCause records the cause the terminal reader ended with. An orderly EOF | |||
| 3941 | + | // is the absence of a failure and stays zero, so cause reports io.EOF for it. | |||
| 3942 | + | func (p *terminalInputPump) setReadCause(err error) { | |||
| 3943 | + | if err == nil || errors.Is(err, io.EOF) { | |||
| 3944 | + | return | |||
| 3945 | + | } | |||
| 3946 | + | p.mu.Lock() | |||
| 3947 | + | if p.readCause == nil { | |||
| 3948 | + | p.readCause = err | |||
| 3949 | + | } | |||
| 3950 | + | p.mu.Unlock() | |||
| 3951 | + | } | |||
| 3952 | + | ||||
| 3953 | + | // cause reports why the terminal read ended: the read failure when there was | |||
| 3954 | + | // one, otherwise an orderly EOF. | |||
| 3955 | + | func (p *terminalInputPump) cause() error { | |||
| 3956 | + | p.mu.Lock() | |||
| 3957 | + | defer p.mu.Unlock() | |||
| 3958 | + | if p.readCause != nil { | |||
| 3959 | + | return p.readCause | |||
| 3960 | + | } | |||
| 3961 | + | return io.EOF | |||
| 3962 | + | } | |||
| 3963 | + | ||||
| 3934 | func (p *terminalInputPump) isClosed() bool { | 3964 | func (p *terminalInputPump) isClosed() bool { | ||
| 3935 | p.mu.Lock() | 3965 | p.mu.Lock() | ||
| 3936 | defer p.mu.Unlock() | 3966 | defer p.mu.Unlock() | ||
M internal/usecase/client/lifecycle_p54a_test.go +18 -0
| @@ -86,6 +86,24 @@ func TestP54AttachmentOutcomesReturnToPicker(t *testing.T) { | |||
| 86 | 86 | } | |
| 87 | 87 | } | |
| 88 | 88 | ||
| 89 | + | func TestP54LocalRefusalIsNotReportedAsDestinationFailure(t *testing.T) { | |
| 90 | + | picker := newAttachTestPicker() | |
| 91 | + | notices := make(chan LifecycleNotice, 4) | |
| 92 | + | harness := startAttachHarness(t, picker) | |
| 93 | + | harness.sup.cfg.NotifyLifecycle = func(notice LifecycleNotice) { notices <- notice } | |
| 94 | + | ||
| 95 | + | // The committed selection is refused locally, before any destination exists: | |
| 96 | + | // reporting that as a destination failure would name the wrong cause. | |
| 97 | + | picker.mu.Lock() | |
| 98 | + | picker.resolveErr = pickerCatalogueError{Code: pickerCatalogueGone, Text: "the committed session is gone"} | |
| 99 | + | picker.mu.Unlock() | |
| 100 | + | picker.commit(sessionTestRequest(true)) | |
| 101 | + | ||
| 102 | + | awaitPickerState(t, harness.sup) | |
| 103 | + | require.Equal(t, LifecycleNoticeSelectionUnavailable, (<-notices).Kind) | |
| 104 | + | require.Empty(t, harness.service.openedRequests(), "a refused selection never dials a destination") | |
| 105 | + | } | |
| 106 | + | ||
| 89 | 107 | func TestP54ResolveInitialUsesRealCatalogueCreation(t *testing.T) { | |
| 90 | 108 | clock := newSupervisorTestClock() | |
| 91 | 109 | catalogue, _ := pickerTestCatalogue(t) | |
| @@ -86,6 +86,24 @@ func TestP54AttachmentOutcomesReturnToPicker(t *testing.T) { | |||||
| 86 | } | 86 | } | ||
| 87 | } | 87 | } | ||
| 88 | 88 | ||||
| 89 | + | func TestP54LocalRefusalIsNotReportedAsDestinationFailure(t *testing.T) { | |||
| 90 | + | picker := newAttachTestPicker() | |||
| 91 | + | notices := make(chan LifecycleNotice, 4) | |||
| 92 | + | harness := startAttachHarness(t, picker) | |||
| 93 | + | harness.sup.cfg.NotifyLifecycle = func(notice LifecycleNotice) { notices <- notice } | |||
| 94 | + | ||||
| 95 | + | // The committed selection is refused locally, before any destination exists: | |||
| 96 | + | // reporting that as a destination failure would name the wrong cause. | |||
| 97 | + | picker.mu.Lock() | |||
| 98 | + | picker.resolveErr = pickerCatalogueError{Code: pickerCatalogueGone, Text: "the committed session is gone"} | |||
| 99 | + | picker.mu.Unlock() | |||
| 100 | + | picker.commit(sessionTestRequest(true)) | |||
| 101 | + | ||||
| 102 | + | awaitPickerState(t, harness.sup) | |||
| 103 | + | require.Equal(t, LifecycleNoticeSelectionUnavailable, (<-notices).Kind) | |||
| 104 | + | require.Empty(t, harness.service.openedRequests(), "a refused selection never dials a destination") | |||
| 105 | + | } | |||
| 106 | + | ||||
| 89 | func TestP54ResolveInitialUsesRealCatalogueCreation(t *testing.T) { | 107 | func TestP54ResolveInitialUsesRealCatalogueCreation(t *testing.T) { | ||
| 90 | clock := newSupervisorTestClock() | 108 | clock := newSupervisorTestClock() | ||
| 91 | catalogue, _ := pickerTestCatalogue(t) | 109 | catalogue, _ := pickerTestCatalogue(t) | ||
M internal/usecase/client/picker_controller.go +4 -2
| @@ -68,8 +68,10 @@ type pickerHost interface { | |||
| 68 | 68 | } | |
| 69 | 69 | ||
| 70 | 70 | // pickerInputConsumer is the supervisor's narrow input seam. ConsumeTerminalRead | |
| 71 | - | // returns true when the picker owned and consumed the read; false leaves the | |
| 72 | - | // read to the supervisor's ordinary (P5.1a) handling. | |
| 71 | + | // returns true when the picker owned and consumed the read. It returns false for | |
| 72 | + | // bytes the picker declined, which are then dropped: the supervisor keeps no | |
| 73 | + | // ordinary terminal handling alongside the picker, so a declined read has no | |
| 74 | + | // second owner to hand it to. | |
| 73 | 75 | type pickerInputConsumer interface { | |
| 74 | 76 | ConsumeTerminalRead(data []byte) bool | |
| 75 | 77 | } | |
| @@ -68,8 +68,10 @@ type pickerHost interface { | |||||
| 68 | } | 68 | } | ||
| 69 | 69 | ||||
| 70 | // pickerInputConsumer is the supervisor's narrow input seam. ConsumeTerminalRead | 70 | // pickerInputConsumer is the supervisor's narrow input seam. ConsumeTerminalRead | ||
| 71 | - | // returns true when the picker owned and consumed the read; false leaves the | 71 | + | // returns true when the picker owned and consumed the read. It returns false for |
| 72 | - | // read to the supervisor's ordinary (P5.1a) handling. | 72 | + | // bytes the picker declined, which are then dropped: the supervisor keeps no |
| 73 | + | // ordinary terminal handling alongside the picker, so a declined read has no | |||
| 74 | + | // second owner to hand it to. | |||
| 73 | type pickerInputConsumer interface { | 75 | type pickerInputConsumer interface { | ||
| 74 | ConsumeTerminalRead(data []byte) bool | 76 | ConsumeTerminalRead(data []byte) bool | ||
| 75 | } | 77 | } | ||
M internal/usecase/client/supervisor.go +10 -1
| @@ -240,6 +240,11 @@ const ( | |||
| 240 | 240 | LifecycleNoticeBrokerLost | |
| 241 | 241 | LifecycleNoticeBrokerReconnected | |
| 242 | 242 | LifecycleNoticeOutcomeUnknown | |
| 243 | + | // LifecycleNoticeSelectionUnavailable reports a local refusal: the picker | |
| 244 | + | // could not resolve the committed selection against the current catalogue, | |
| 245 | + | // or initial navigation had nothing to resolve. No destination was dialed, | |
| 246 | + | // so reporting it as a destination failure would name the wrong cause. | |
| 247 | + | LifecycleNoticeSelectionUnavailable | |
| 243 | 248 | ) | |
| 244 | 249 | ||
| 245 | 250 | // LifecycleNotice carries classification only; adapter diagnostics and free | |
| @@ -956,7 +961,11 @@ func startTerminalInputLifetime(in io.Reader, consumer pickerInputConsumer) *ter | |||
| 956 | 961 | go lifetime.runPicker() | |
| 957 | 962 | go func() { | |
| 958 | 963 | <-lifetime.pump.exited | |
| 959 | - | lifetime.finish(io.EOF) | |
| 964 | + | // Publish the cause the reader actually ended with. The reader enqueues | |
| 965 | + | // its final result for the consumer, but this watcher can win that race, | |
| 966 | + | // and reporting a read failure as an orderly EOF would return a clean | |
| 967 | + | // status for a terminal that failed. | |
| 968 | + | lifetime.finish(lifetime.pump.cause()) | |
| 960 | 969 | }() | |
| 961 | 970 | return lifetime | |
| 962 | 971 | } | |
| @@ -240,6 +240,11 @@ const ( | |||||
| 240 | LifecycleNoticeBrokerLost | 240 | LifecycleNoticeBrokerLost | ||
| 241 | LifecycleNoticeBrokerReconnected | 241 | LifecycleNoticeBrokerReconnected | ||
| 242 | LifecycleNoticeOutcomeUnknown | 242 | LifecycleNoticeOutcomeUnknown | ||
| 243 | + | // LifecycleNoticeSelectionUnavailable reports a local refusal: the picker | |||
| 244 | + | // could not resolve the committed selection against the current catalogue, | |||
| 245 | + | // or initial navigation had nothing to resolve. No destination was dialed, | |||
| 246 | + | // so reporting it as a destination failure would name the wrong cause. | |||
| 247 | + | LifecycleNoticeSelectionUnavailable | |||
| 243 | ) | 248 | ) | ||
| 244 | 249 | ||||
| 245 | // LifecycleNotice carries classification only; adapter diagnostics and free | 250 | // LifecycleNotice carries classification only; adapter diagnostics and free | ||
| @@ -956,7 +961,11 @@ func startTerminalInputLifetime(in io.Reader, consumer pickerInputConsumer) *ter | |||||
| 956 | go lifetime.runPicker() | 961 | go lifetime.runPicker() | ||
| 957 | go func() { | 962 | go func() { | ||
| 958 | <-lifetime.pump.exited | 963 | <-lifetime.pump.exited | ||
| 959 | - | lifetime.finish(io.EOF) | 964 | + | // Publish the cause the reader actually ended with. The reader enqueues |
| 965 | + | // its final result for the consumer, but this watcher can win that race, | |||
| 966 | + | // and reporting a read failure as an orderly EOF would return a clean | |||
| 967 | + | // status for a terminal that failed. | |||
| 968 | + | lifetime.finish(lifetime.pump.cause()) | |||
| 960 | }() | 969 | }() | ||
| 961 | return lifetime | 970 | return lifetime | ||
| 962 | } | 971 | } | ||
M internal/usecase/client/supervisor_attach.go +9 -2
| @@ -432,10 +432,17 @@ func (s *Supervisor) settleAttachment(ctx context.Context, input *terminalInputL | |||
| 432 | 432 | // reportAttachmentFailure returns to the picker with a typed, visible failure. | |
| 433 | 433 | func (s *Supervisor) reportAttachmentFailure(err error) { | |
| 434 | 434 | s.transition(supervisorEvent{kind: supervisorAttachEnded, err: err}) | |
| 435 | + | // Classification order matters: a local refusal never dialed a destination, | |
| 436 | + | // so it must not be reported as one, and only a genuinely unknown outcome | |
| 437 | + | // gets its own notice. | |
| 438 | + | var catalogueErr pickerCatalogueError | |
| 435 | 439 | var brokerErr ports.BrokerError | |
| 436 | - | if errors.As(err, &brokerErr) && brokerErr.Code == ports.BrokerErrorOutcomeUnknown { | |
| 440 | + | switch { | |
| 441 | + | case errors.As(err, &catalogueErr): | |
| 442 | + | s.notifyLifecycle(LifecycleNoticeSelectionUnavailable) | |
| 443 | + | case errors.As(err, &brokerErr) && brokerErr.Code == ports.BrokerErrorOutcomeUnknown: | |
| 437 | 444 | s.notifyLifecycle(LifecycleNoticeOutcomeUnknown) | |
| 438 | - | } else { | |
| 445 | + | default: | |
| 439 | 446 | s.notifyLifecycle(LifecycleNoticeDestinationFailed) | |
| 440 | 447 | } | |
| 441 | 448 | s.notifyAttachment(err) | |
| @@ -432,10 +432,17 @@ func (s *Supervisor) settleAttachment(ctx context.Context, input *terminalInputL | |||||
| 432 | // reportAttachmentFailure returns to the picker with a typed, visible failure. | 432 | // reportAttachmentFailure returns to the picker with a typed, visible failure. | ||
| 433 | func (s *Supervisor) reportAttachmentFailure(err error) { | 433 | func (s *Supervisor) reportAttachmentFailure(err error) { | ||
| 434 | s.transition(supervisorEvent{kind: supervisorAttachEnded, err: err}) | 434 | s.transition(supervisorEvent{kind: supervisorAttachEnded, err: err}) | ||
| 435 | + | // Classification order matters: a local refusal never dialed a destination, | |||
| 436 | + | // so it must not be reported as one, and only a genuinely unknown outcome | |||
| 437 | + | // gets its own notice. | |||
| 438 | + | var catalogueErr pickerCatalogueError | |||
| 435 | var brokerErr ports.BrokerError | 439 | var brokerErr ports.BrokerError | ||
| 436 | - | if errors.As(err, &brokerErr) && brokerErr.Code == ports.BrokerErrorOutcomeUnknown { | 440 | + | switch { |
| 441 | + | case errors.As(err, &catalogueErr): | |||
| 442 | + | s.notifyLifecycle(LifecycleNoticeSelectionUnavailable) | |||
| 443 | + | case errors.As(err, &brokerErr) && brokerErr.Code == ports.BrokerErrorOutcomeUnknown: | |||
| 437 | s.notifyLifecycle(LifecycleNoticeOutcomeUnknown) | 444 | s.notifyLifecycle(LifecycleNoticeOutcomeUnknown) | ||
| 438 | - | } else { | 445 | + | default: |
| 439 | s.notifyLifecycle(LifecycleNoticeDestinationFailed) | 446 | s.notifyLifecycle(LifecycleNoticeDestinationFailed) | ||
| 440 | } | 447 | } | ||
| 441 | s.notifyAttachment(err) | 448 | s.notifyAttachment(err) | ||
M internal/usecase/client/supervisor_attach_test.go +8 -2
| @@ -138,8 +138,11 @@ type attachTestPicker struct { | |||
| 138 | 138 | consumed int | |
| 139 | 139 | op pickerOp | |
| 140 | 140 | key string | |
| 141 | - | request ports.BrokerOpenStreamRequest | |
| 142 | - | resolved int | |
| 141 | + | // resolveErr makes ResolveKey refuse locally, which is how a test reaches | |
| 142 | + | // the local-refusal classification without dialing anything. | |
| 143 | + | resolveErr error | |
| 144 | + | request ports.BrokerOpenStreamRequest | |
| 145 | + | resolved int | |
| 143 | 146 | } | |
| 144 | 147 | ||
| 145 | 148 | func newAttachTestPicker() *attachTestPicker { | |
| @@ -186,6 +189,9 @@ func (p *attachTestPicker) ResolveKey(key string, base pickerResolveBase) (ports | |||
| 186 | 189 | p.mu.Lock() | |
| 187 | 190 | defer p.mu.Unlock() | |
| 188 | 191 | p.resolved++ | |
| 192 | + | if p.resolveErr != nil { | |
| 193 | + | return ports.BrokerOpenStreamRequest{}, p.resolveErr | |
| 194 | + | } | |
| 189 | 195 | request := p.request | |
| 190 | 196 | request.Connection = base.Connection | |
| 191 | 197 | request.Stream = base.Stream | |
| @@ -138,8 +138,11 @@ type attachTestPicker struct { | |||||
| 138 | consumed int | 138 | consumed int | ||
| 139 | op pickerOp | 139 | op pickerOp | ||
| 140 | key string | 140 | key string | ||
| 141 | - | request ports.BrokerOpenStreamRequest | 141 | + | // resolveErr makes ResolveKey refuse locally, which is how a test reaches |
| 142 | - | resolved int | 142 | + | // the local-refusal classification without dialing anything. |
| 143 | + | resolveErr error | |||
| 144 | + | request ports.BrokerOpenStreamRequest | |||
| 145 | + | resolved int | |||
| 143 | } | 146 | } | ||
| 144 | 147 | ||||
| 145 | func newAttachTestPicker() *attachTestPicker { | 148 | func newAttachTestPicker() *attachTestPicker { | ||
| @@ -186,6 +189,9 @@ func (p *attachTestPicker) ResolveKey(key string, base pickerResolveBase) (ports | |||||
| 186 | p.mu.Lock() | 189 | p.mu.Lock() | ||
| 187 | defer p.mu.Unlock() | 190 | defer p.mu.Unlock() | ||
| 188 | p.resolved++ | 191 | p.resolved++ | ||
| 192 | + | if p.resolveErr != nil { | |||
| 193 | + | return ports.BrokerOpenStreamRequest{}, p.resolveErr | |||
| 194 | + | } | |||
| 189 | request := p.request | 195 | request := p.request | ||
| 190 | request.Connection = base.Connection | 196 | request.Connection = base.Connection | ||
| 191 | request.Stream = base.Stream | 197 | request.Stream = base.Stream | ||
A internal/usecase/client/terminal_cause_test.go +50 -0
| @@ -0,0 +1,50 @@ | |||
| 1 | + | package client | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "errors" | |
| 5 | + | "io" | |
| 6 | + | "testing" | |
| 7 | + | "time" | |
| 8 | + | ||
| 9 | + | "github.com/stretchr/testify/require" | |
| 10 | + | ) | |
| 11 | + | ||
| 12 | + | // failingReader ends the terminal read with a fixed cause. | |
| 13 | + | type failingReader struct{ err error } | |
| 14 | + | ||
| 15 | + | func (r failingReader) Read([]byte) (int, error) { return 0, r.err } | |
| 16 | + | ||
| 17 | + | // TestTerminalReadCauseIsReportedExactly proves the input lifetime publishes the | |
| 18 | + | // cause the terminal reader ended with rather than a hardcoded EOF. | |
| 19 | + | // | |
| 20 | + | // A read failure reported as an orderly EOF makes the supervisor return a clean | |
| 21 | + | // status for a terminal that failed. The reader enqueues its final result for a | |
| 22 | + | // consumer to report, but the exit watcher can win that race, so the pump | |
| 23 | + | // records the cause and the watcher publishes it. The nil consumer here leaves | |
| 24 | + | // that result unconsumed, which is exactly the racing shape. | |
| 25 | + | func TestTerminalReadCauseIsReportedExactly(t *testing.T) { | |
| 26 | + | failure := errors.New("terminal read failed") | |
| 27 | + | ||
| 28 | + | tests := []struct { | |
| 29 | + | name string | |
| 30 | + | reader io.Reader | |
| 31 | + | want error | |
| 32 | + | }{ | |
| 33 | + | {name: "read failure is not an orderly exit", reader: failingReader{err: failure}, want: failure}, | |
| 34 | + | {name: "orderly end of input stays eof", reader: failingReader{err: io.EOF}, want: io.EOF}, | |
| 35 | + | } | |
| 36 | + | ||
| 37 | + | for _, tc := range tests { | |
| 38 | + | t.Run(tc.name, func(t *testing.T) { | |
| 39 | + | lifetime := startTerminalInputLifetime(tc.reader, nil) | |
| 40 | + | t.Cleanup(lifetime.stop) | |
| 41 | + | ||
| 42 | + | select { | |
| 43 | + | case err := <-lifetime.EOF(): | |
| 44 | + | require.ErrorIs(t, err, tc.want) | |
| 45 | + | case <-time.After(5 * time.Second): | |
| 46 | + | t.Fatal("terminal cause was never published") | |
| 47 | + | } | |
| 48 | + | }) | |
| 49 | + | } | |
| 50 | + | } | |
| @@ -0,0 +1,50 @@ | |||||
| 1 | + | package client | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "errors" | |||
| 5 | + | "io" | |||
| 6 | + | "testing" | |||
| 7 | + | "time" | |||
| 8 | + | ||||
| 9 | + | "github.com/stretchr/testify/require" | |||
| 10 | + | ) | |||
| 11 | + | ||||
| 12 | + | // failingReader ends the terminal read with a fixed cause. | |||
| 13 | + | type failingReader struct{ err error } | |||
| 14 | + | ||||
| 15 | + | func (r failingReader) Read([]byte) (int, error) { return 0, r.err } | |||
| 16 | + | ||||
| 17 | + | // TestTerminalReadCauseIsReportedExactly proves the input lifetime publishes the | |||
| 18 | + | // cause the terminal reader ended with rather than a hardcoded EOF. | |||
| 19 | + | // | |||
| 20 | + | // A read failure reported as an orderly EOF makes the supervisor return a clean | |||
| 21 | + | // status for a terminal that failed. The reader enqueues its final result for a | |||
| 22 | + | // consumer to report, but the exit watcher can win that race, so the pump | |||
| 23 | + | // records the cause and the watcher publishes it. The nil consumer here leaves | |||
| 24 | + | // that result unconsumed, which is exactly the racing shape. | |||
| 25 | + | func TestTerminalReadCauseIsReportedExactly(t *testing.T) { | |||
| 26 | + | failure := errors.New("terminal read failed") | |||
| 27 | + | ||||
| 28 | + | tests := []struct { | |||
| 29 | + | name string | |||
| 30 | + | reader io.Reader | |||
| 31 | + | want error | |||
| 32 | + | }{ | |||
| 33 | + | {name: "read failure is not an orderly exit", reader: failingReader{err: failure}, want: failure}, | |||
| 34 | + | {name: "orderly end of input stays eof", reader: failingReader{err: io.EOF}, want: io.EOF}, | |||
| 35 | + | } | |||
| 36 | + | ||||
| 37 | + | for _, tc := range tests { | |||
| 38 | + | t.Run(tc.name, func(t *testing.T) { | |||
| 39 | + | lifetime := startTerminalInputLifetime(tc.reader, nil) | |||
| 40 | + | t.Cleanup(lifetime.stop) | |||
| 41 | + | ||||
| 42 | + | select { | |||
| 43 | + | case err := <-lifetime.EOF(): | |||
| 44 | + | require.ErrorIs(t, err, tc.want) | |||
| 45 | + | case <-time.After(5 * time.Second): | |||
| 46 | + | t.Fatal("terminal cause was never published") | |||
| 47 | + | } | |||
| 48 | + | }) | |||
| 49 | + | } | |||
| 50 | + | } | |||