diff --git a/AGENTS.md b/AGENTS.md index ea4d29d..db00f8a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -33,7 +33,8 @@ the opposite caused a bug that took hours to find. 3. **Polling lives in `tea.Cmd`.** A poll returns a message. A goroutine never writes to the model, never draws, and never blocks sending on a channel the UI also reads. A timer starts only through `startTimer` - (`internal/ui/timer.go`). + (`internal/ui/timer.go`). A command waits on a stream only through `waitOn` + (`internal/ui/stream.go`). 4. **Events say when, reads say what.** The event stream marks a screen stale, and the screen is read again the usual way. A cluster that does not stream is polled as before. Event payloads are read in two places only: diff --git a/internal/ui/feed.go b/internal/ui/feed.go index 3093a19..7c3bb5c 100644 --- a/internal/ui/feed.go +++ b/internal/ui/feed.go @@ -200,19 +200,15 @@ func nextEvent(feed *nomad.Feed) tea.Cmd { return nil } - return func() tea.Msg { - select { - case event, ok := <-feed.C: - if !ok { - return feedEndedMsg{feed: feed, err: reasonOf(feed)} - } - - return feedEventsMsg{feed: feed, events: withWaiting(feed, event)} - - case err := <-feed.Err: - return feedEndedMsg{feed: feed, err: err} + return waitOn(feed.C, feed.Err, func(event nomad.Event, ok bool) tea.Msg { + if !ok { + return feedEndedMsg{feed: feed, err: reasonOf(feed)} } - } + + return feedEventsMsg{feed: feed, events: withWaiting(feed, event)} + }, func(err error) tea.Msg { + return feedEndedMsg{feed: feed, err: err} + }) } // withWaiting is the first event and every one waiting behind it, oldest diff --git a/internal/ui/logs.go b/internal/ui/logs.go index d6d7e3d..7f5490c 100644 --- a/internal/ui/logs.go +++ b/internal/ui/logs.go @@ -256,23 +256,19 @@ func waitForStream(stream *nomad.LogStream) tea.Cmd { return nil } - return func() tea.Msg { - select { - case line, ok := <-stream.Lines: - if !ok { - return endOf(stream) - } - - return logLineMsg{stream: stream, text: line, start: stream.Start(), old: stream.Old(line)} - - case err := <-stream.Err: - if err != nil { - return errMsg{err: err} - } + return waitOn(stream.Lines, stream.Err, func(line string, ok bool) tea.Msg { + if !ok { + return endOf(stream) + } - return logEndMsg{stream: stream} + return logLineMsg{stream: stream, text: line, start: stream.Start(), old: stream.Old(line)} + }, func(err error) tea.Msg { + if err != nil { + return errMsg{err: err} } - } + + return logEndMsg{stream: stream} + }) } // endOf is the end of a stream whose lines closed: its error, when it sent diff --git a/internal/ui/stream.go b/internal/ui/stream.go new file mode 100644 index 0000000..0c2b421 --- /dev/null +++ b/internal/ui/stream.go @@ -0,0 +1,39 @@ +package ui + +import ( + "sync/atomic" + + tea "charm.land/bubbletea/v2" +) + +// heldStream is what a held stream wait answers when nothing has come: it +// would wait. +type heldStream struct{} + +// streamsHeld is set by tests that play out commands, to tell a stream wait +// from slow work. +var streamsHeld atomic.Bool + +// waitOn is the one place a command waits on a stream: for the next thing +// on c, or for the error the stream ends with. A held wait does not wait. +func waitOn[T any](c <-chan T, errs <-chan error, next func(T, bool) tea.Msg, ended func(error) tea.Msg) tea.Cmd { + return func() tea.Msg { + if streamsHeld.Load() { + select { + case v, ok := <-c: + return next(v, ok) + case err := <-errs: + return ended(err) + default: + return heldStream{} + } + } + + select { + case v, ok := <-c: + return next(v, ok) + case err := <-errs: + return ended(err) + } + } +} diff --git a/internal/ui/stream_test.go b/internal/ui/stream_test.go new file mode 100644 index 0000000..7192eb7 --- /dev/null +++ b/internal/ui/stream_test.go @@ -0,0 +1,119 @@ +package ui + +import ( + "go/ast" + "go/token" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/ingvarch/urga/internal/nomad" +) + +// waitsOnChannels are the functions that may wait on a channel: the one +// place a command waits on a stream, and the shell's watch of the terminal +// size, which runs in the shell's own goroutine and not in a command. +var waitsOnChannels = map[string]bool{"waitOn": true, "watchSize": true, "sendSize": true} + +func TestStreams_AreWaitedOnInOnePlace(t *testing.T) { + fset, files := sourceFiles(t) + + for _, file := range files { + for _, decl := range file.Decls { + fn, ok := decl.(*ast.FuncDecl) + if !ok || fn.Body == nil || waitsOnChannels[fn.Name.Name] { + continue + } + + for _, at := range channelWaits(fn.Body) { + t.Errorf("%s waits on a channel outside waitOn: %s", fn.Name.Name, fset.Position(at)) + } + } + } +} + +// channelWaits are the receives in body that can block: every receive but +// the cases of a select that has a default. A range over a channel waits too, +// but finding one needs the types; playOut names such a command when it +// does not answer. +func channelWaits(body *ast.BlockStmt) []token.Pos { + nonBlocking := map[ast.Node]bool{} + + ast.Inspect(body, func(n ast.Node) bool { + sel, ok := n.(*ast.SelectStmt) + if !ok || !hasDefault(sel) { + return true + } + + for _, stmt := range sel.Body.List { + if comm := stmt.(*ast.CommClause).Comm; comm != nil { + ast.Inspect(comm, func(n ast.Node) bool { + if recv, ok := n.(*ast.UnaryExpr); ok && recv.Op == token.ARROW { + nonBlocking[recv] = true + } + + return true + }) + } + } + + return true + }) + + var waits []token.Pos + + ast.Inspect(body, func(n ast.Node) bool { + if recv, ok := n.(*ast.UnaryExpr); ok && recv.Op == token.ARROW && !nonBlocking[recv] { + waits = append(waits, recv.Pos()) + } + + return true + }) + + return waits +} + +func hasDefault(sel *ast.SelectStmt) bool { + for _, stmt := range sel.Body.List { + if stmt.(*ast.CommClause).Comm == nil { + return true + } + } + + return false +} + +// holdStreams makes every stream wait answer at once, for the length of the +// test. +func holdStreams(t *testing.T) { + t.Helper() + + held := streamsHeld.Swap(true) + + t.Cleanup(func() { streamsHeld.Store(held) }) +} + +func TestStreams_AHeldWaitAnswersWithWhatCameOrAtOnce(t *testing.T) { + r := require.New(t) + + holdStreams(t) + + lines := make(chan string, 1) + lines <- "the task says something\n" + stream := &nomad.LogStream{Lines: lines, Err: make(chan error)} + + msg, ok := within(waitForStream(stream), time.Second).(logLineMsg) + r.True(ok) + r.Equal("the task says something\n", msg.text) + + // Nothing more came: the wait answers that it would wait. + r.Equal(heldStream{}, within(waitForStream(stream), time.Second)) +} + +func TestPlayOut_LeavesAStreamThatSentNothing(t *testing.T) { + m := playOut(loadedModel(t), waitForStream(&nomad.LogStream{Lines: make(chan string)})) + + require.Contains(t, plain(m.render()), "web") + require.False(t, streamsHeld.Load()) +} diff --git a/internal/ui/watch.go b/internal/ui/watch.go index 6689ec0..0d5ea51 100644 --- a/internal/ui/watch.go +++ b/internal/ui/watch.go @@ -107,22 +107,18 @@ func (w watchState) waitForChange() tea.Cmd { id := w.id - return func() tea.Msg { - select { - case _, ok := <-changes.C: - if !ok { - // A stream that dropped sends its error before it - // closes, and both are then ready at once: reading the - // close first must not lose the error. - return watchEndedMsg{id: id, err: reasonOf(changes)} - } - - return changeMsg{id: id} - - case err := <-changes.Err: - return watchEndedMsg{id: id, err: err} + return waitOn(changes.C, changes.Err, func(_ nomad.Change, ok bool) tea.Msg { + if !ok { + // A stream that dropped sends its error before it + // closes, and both are then ready at once: reading the + // close first must not lose the error. + return watchEndedMsg{id: id, err: reasonOf(changes)} } - } + + return changeMsg{id: id} + }, func(err error) tea.Msg { + return watchEndedMsg{id: id, err: err} + }) } // reasonOf is the error a stream ended with, if it sent one. diff --git a/internal/ui/writes_test.go b/internal/ui/writes_test.go index 764a668..1f27c2c 100644 --- a/internal/ui/writes_test.go +++ b/internal/ui/writes_test.go @@ -3,6 +3,7 @@ package ui import ( "fmt" "reflect" + "runtime" "slices" "testing" "time" @@ -88,122 +89,85 @@ func TestFake_RecordsEveryWrite(t *testing.T) { } // playOut runs what a key started the way the program does, and answers -// what it asks the way someone who means it would. A command that has not -// answered after a moment keeps running, and its answer is taken when it -// comes: a loaded machine can be slow. Timers are held while it runs, so a -// timer answers at once and nothing takes it. One still waiting when the rest -// is done, like a stream, is left waiting. +// what it asks the way someone who means it would. Timers and stream waits +// are held while it runs, so each answers at once: a timer with what it +// would send, which nothing takes, and a stream wait with what came, or +// with nothing when nothing came. Every other command is waited for until +// it answers. func playOut(m Model, cmd tea.Cmd) Model { defer timersHeld.Store(timersHeld.Swap(true)) - queue := []tea.Cmd{cmd} - - var running []runningCmd + defer streamsHeld.Store(streamsHeld.Swap(true)) - take := func(msg tea.Msg) { - switch msg := msg.(type) { - case nil: - case tea.BatchMsg: - queue = append(queue, msg...) - default: - var next tea.Cmd - m, next = m.update(msg) - queue = append(queue, next) - } - } + queue := []tea.Cmd{cmd} // A plan is answered when the key opened it. One that was on the screen // before the key is not sent. _, planned := m.screen.page.(planPage) for range 300 { - running = takeAnswered(running, take) - if len(queue) == 0 { if _, onPlan := m.screen.page.(planPage); planned && onPlan && m.overlay == overlayNone { return m } next, answered, acted := answer(m) - if acted { - m, queue = next, append(queue, answered) - - continue + if !acted { + return m } - if running = waitForRecent(running, take); len(queue) > 0 { - continue - } + m, queue = next, append(queue, answered) - return m + continue } run := queue[0] queue = queue[1:] - if run == nil { - continue - } - - started := runningCmd{answer: make(chan tea.Msg, 1), at: time.Now()} - go func() { started.answer <- run() }() - - select { - case msg := <-started.answer: - take(msg) - case <-time.After(50 * time.Millisecond): - running = append(running, started) + switch msg := answerOf(run, slowest).(type) { + case nil, heldStream: + case tea.BatchMsg: + queue = append(queue, msg...) + default: + var next tea.Cmd + m, next = m.update(msg) + queue = append(queue, next) } } return m } -// runningCmd is a command that has not answered yet, and when it started. -type runningCmd struct { - answer chan tea.Msg - at time.Time -} - -// lateAnswer is how long a command may take, started just before the rest -// is done, before it is left waiting. -const lateAnswer = 100 * time.Millisecond - -// takeAnswered takes the answers that came, and keeps the commands that -// have none yet. -func takeAnswered(running []runningCmd, take func(tea.Msg)) []runningCmd { - still := running[:0] +// slowest is how long playOut waits for a command to answer. One that takes +// longer panics with the name of the command, which ends the run of the +// package; the test would otherwise hang. +const slowest = 5 * time.Second - for _, r := range running { - select { - case msg := <-r.answer: - take(msg) - default: - still = append(still, r) - } +// answerOf is what cmd answers. It panics when cmd takes longer than wait. +func answerOf(cmd tea.Cmd, wait time.Duration) tea.Msg { + msg, ok := answerIn(cmd, wait) + if !ok { + panic(fmt.Sprintf("%s did not answer in %s", runtime.FuncForPC(reflect.ValueOf(cmd).Pointer()).Name(), wait)) } - return still + return msg } -// waitForRecent waits for an answer of a command started less than -// lateAnswer ago, and takes the first that comes. -func waitForRecent(running []runningCmd, take func(tea.Msg)) []runningCmd { - var until time.Time - for _, r := range running { - if end := r.at.Add(lateAnswer); end.After(until) { - until = end - } +// answerIn is what cmd answers, and false when it takes longer than wait. +func answerIn(cmd tea.Cmd, wait time.Duration) (tea.Msg, bool) { + if cmd == nil { + return nil, true } - for time.Now().Before(until) { - if still := takeAnswered(running, take); len(still) < len(running) { - return still - } + answered := make(chan tea.Msg, 1) - time.Sleep(2 * time.Millisecond) - } + go func() { answered <- cmd() }() - return running + select { + case msg := <-answered: + return msg, true + case <-time.After(wait): + return nil, false + } } func TestPlayOut_TakesAnAnswerThatComesLate(t *testing.T) { @@ -221,6 +185,34 @@ func TestPlayOut_TakesAnAnswerThatComesLate(t *testing.T) { require.Contains(t, plain(m.render()), "the answer came late") } +func TestPlayOut_WaitsForSlowWorkUntilItAnswers(t *testing.T) { + m := loadedModel(t) + + slow := func() tea.Msg { + time.Sleep(300 * time.Millisecond) + + return sayMsg("the slow work is done") + } + + m = playOut(m, slow) + + require.Contains(t, plain(m.render()), "the slow work is done") +} + +func TestAnswerOf_PanicsOnWorkThatNeverAnswers(t *testing.T) { + never := make(chan struct{}) + t.Cleanup(func() { close(never) }) + + stuck := func() tea.Msg { + <-never + + return nil + } + + require.PanicsWithValue(t, runtime.FuncForPC(reflect.ValueOf(stuck).Pointer()).Name()+" did not answer in 10ms", + func() { answerOf(stuck, 10*time.Millisecond) }) +} + // answer says yes to a question and to a plan, and answers a line with what // it offers, or with a word when it offers nothing. func answer(m Model) (Model, tea.Cmd, bool) { @@ -251,20 +243,9 @@ func answer(m Model) (Model, tea.Cmd, bool) { // within is what a command answers, unless it takes longer than a moment. func within(cmd tea.Cmd, wait time.Duration) tea.Msg { - if cmd == nil { - return nil - } + msg, _ := answerIn(cmd, wait) - answered := make(chan tea.Msg, 1) - - go func() { answered <- cmd() }() - - select { - case msg := <-answered: - return msg - case <-time.After(wait): - return nil - } + return msg } func TestBindings_TheKeysThatChangeTheCluster(t *testing.T) {