Commit `3ac40e6851`

3ac40e6851598d8213bf09677398b25a4d6cbf6a

parent: be53d945d3

Verified · cmc

cmc <hello@cleberg.net> · 2026-09-23 06:20 UTC

build log --follow: end a queued follow, stream only on GET

A build on a repository with no runner stayed pending forever; nothing
reaps it and a follow of it never ended. followBuildLog now tracks how
long the build has been pending during the follow and gives up after
followQueued (10m), writing to stderr and exiting ExitFailure.

The web build page streamed on any method matching the route; a HEAD
now renders once instead. After the command returns, streamBuild
writes nothing more once the client has left, uses a signed-out-scoped
message for the follow-cap denial, and surfaces an ExitFailure message
(the queued limit) as a notice paragraph. htmlStream.Write escapes into
a buffer and returns the write error instead of discarding it;
gzipWriter gains FlushError so a failed flush reaches the caller.

TestBuildLogFollowDone now appends a line and waits for the follower to
read it before closing Done, so it proves an in-progress wait is
interrupted rather than a follow that never blocked.

Ref #250

internal/control/buildfollow.go +22

		@@ -22,6 +22,12 @@ var (
22	22	// an outcome: a cancel appends its line after the status changes, and
23	23	// a cancelled runner's stream runs on until its next check.
24	24	followSettle = time.Second
	25	// followQueued bounds how long a follow waits on a build that stays
	26	// pending: nothing reaps a queued build (ReapStaleBuilds only reaps
	27	// running builds), and a running one is already bounded by the
	28	// reaper's deadline, so a follow needs its own limit for the queued
	29	// case or it never ends.
	30	followQueued = 10 * time.Minute
25	31	)
26	32
27	33	var (
		@@ -58,6 +64,7 @@ func followBuildLog(c *Ctx, b store.Build) int {
58	64
59	65	var off int64
60	66	var settleBy time.Time
	67	var queuedSince time.Time
61	68	for {
62	69	wake := c.Store.BuildLogWait(b.ID)
63	70	status, chunk, err := c.Store.BuildLogFrom(b.ID, off)
		@@ -70,7 +77,22 @@ func followBuildLog(c *Ctx, b store.Build) int {
70	77	}
71	78	off += int64(len(chunk))
72	79	}
	80	if status == "pending" {
	81	if queuedSince.IsZero() {
	82	queuedSince = time.Now()
	83	}
	84	} else {
	85	queuedSince = time.Time{}
	86	}
73	87	wait := followPoll
	88	if status == "pending" {
	89	left := queuedSince.Add(followQueued).Sub(time.Now())
	90	if left <= 0 {
	91	fmt.Fprintf(c.Stderr, "build %d is still queued; nothing claimed it in %v. Follow again once a runner has.\n", b.Number, followQueued)
	92	return protocol.ExitFailure
	93	}
	94	wait = min(wait, left)
	95	}
74	96	if status != "pending" && status != "running" {
75	97	if settleBy.IsZero() {
76	98	settleBy = time.Now().Add(followSettle)

internal/control/buildfollow_test.go +59 −7

		@@ -3,6 +3,7 @@ package control
3	3	import (
4	4	"bytes"
5	5	"strings"
	6	"sync"
6	7	"testing"
7	8	"time"
8	9
		@@ -10,15 +11,34 @@ import (
10	11	"gitbay.org/gitbay/internal/store"
11	12	)
12	13
	14	// syncBuffer is a bytes.Buffer guarded by a mutex, safe for a test to poll
	15	// while the follow goroutine is still writing to it.
	16	type syncBuffer struct {
	17	mu sync.Mutex
	18	buf bytes.Buffer
	19	}
	20
	21	func (b *syncBuffer) Write(p []byte) (int, error) {
	22	b.mu.Lock()
	23	defer b.mu.Unlock()
	24	return b.buf.Write(p)
	25	}
	26
	27	func (b *syncBuffer) String() string {
	28	b.mu.Lock()
	29	defer b.mu.Unlock()
	30	return b.buf.String()
	31	}
	32
13	33	// follow starts build log --follow on build 1 of repo and returns the
14	34	// buffers and a channel carrying the exit code.
15		func follow(t testing.T, st store.Store, uid int64, repo store.Repo, done <-chan struct{}) (bytes.Buffer, bytes.Buffer, chan int) {
	35	func follow(t testing.T, st store.Store, uid int64, repo store.Repo, done <-chan struct{}) (syncBuffer, syncBuffer, chan int) {
16	36	t.Helper()
17	37	u, err := st.UserByID(uid)
18	38	if err != nil {
19	39	t.Fatal(err)
20	40	}
21		var out, errOut bytes.Buffer
	41	var out, errOut syncBuffer
22	42	c := &Ctx{User: u, Scope: "full", Store: st, Stdin: strings.NewReader(""),
23	43	Stdout: &out, Stderr: &errOut, Done: done}
24	44	res := make(chan int, 1)
		@@ -38,9 +58,9 @@ func waitExit(t *testing.T, res chan int) int {
38	58	}
39	59
40	60	func shortFollowTimers(t *testing.T) {
41		settle, poll := followSettle, followPoll
	61	settle, poll, queued := followSettle, followPoll, followQueued
42	62	followSettle, followPoll = 200time.Millisecond, 50time.Millisecond
43		t.Cleanup(func() { followSettle, followPoll = settle, poll })
	63	t.Cleanup(func() { followSettle, followPoll, followQueued = settle, poll, queued })
44	64	}
45	65
46	66	// The follow prints the stored log, then what arrives, and ends with the
		@@ -99,21 +119,53 @@ func TestBuildLogFollowCancel(t *testing.T) {
99	119	}
100	120	}
101	121
102		// Closing Done ends a follow of a build that is still running.
	122	// Closing Done ends a follow of a build that is still running, even while
	123	// it is blocked waiting for the next change: the build gets a line, the
	124	// follower is confirmed to have read it (so it is back in its wait), then
	125	// Done closes.
103	126	func TestBuildLogFollowDone(t *testing.T) {
104	127	shortFollowTimers(t)
105	128	st, repo, uid := newQueueTestRepo(t)
106		if _, err := st.CreateBuild(repo.ID, "unit", "abc", "main", `["true"]`, "", "", true); err != nil {
	129	id, err := st.CreateBuild(repo.ID, "unit", "abc", "main", `["true"]`, "", "", true)
	130	if err != nil {
107	131	t.Fatal(err)
108	132	}
	133	st.AppendBuildLog(id, []byte("step one\n"))
109	134	done := make(chan struct{})
110		_, _, res := follow(t, st, uid, repo, done)
	135	out, _, res := follow(t, st, uid, repo, done)
	136
	137	deadline := time.After(2 * time.Second)
	138	for !strings.Contains(out.String(), "step one") {
	139	select {
	140	case <-deadline:
	141	t.Fatal("follow never read the appended line")
	142	case <-time.After(10 * time.Millisecond):
	143	}
	144	}
111	145	close(done)
112	146	if code := waitExit(t, res); code != protocol.ExitFailure {
113	147	t.Fatalf("exit %d, want %d", code, protocol.ExitFailure)
114	148	}
115	149	}
116	150
	151	// A build that stays pending ends its own follow: nothing reaps a queued
	152	// build, so the follow must give up on its own.
	153	func TestBuildLogFollowQueued(t *testing.T) {
	154	shortFollowTimers(t)
	155	followQueued = 150 * time.Millisecond
	156	st, repo, uid := newQueueTestRepo(t)
	157	if _, err := st.CreateBuild(repo.ID, "unit", "abc", "main", `["true"]`, "", "", true); err != nil {
	158	t.Fatal(err)
	159	}
	160	_, errOut, res := follow(t, st, uid, repo, nil)
	161	if code := waitExit(t, res); code != protocol.ExitFailure {
	162	t.Fatalf("exit %d, want %d: %s", code, protocol.ExitFailure, errOut)
	163	}
	164	if !strings.Contains(errOut.String(), "still queued") {
	165	t.Errorf("stderr %q", errOut)
	166	}
	167	}
	168
117	169	// An account holding maxFollows is refused another.
118	170	func TestBuildLogFollowCap(t *testing.T) {
119	171	st, repo, uid := newQueueTestRepo(t)

internal/httpd/builds.go +15 −2

		@@ -304,7 +304,7 @@ func (s Server) build(w http.ResponseWriter, r http.Request) {
304	304	return
305	305	}
306	306	v := buildView{repoPage: p, Build: b, CanWrite: s.canWriteRepo(r, p.Repo), Notice: s.takeFlash(w, r)}
307		if (b.Status == "pending" \|\| b.Status == "running") && r.URL.Query().Get("follow") != "0" {
	307	if (b.Status == "pending" \|\| b.Status == "running") && r.URL.Query().Get("follow") != "0" && r.Method == http.MethodGet {
308	308	s.streamBuild(w, r, v, viewer, n)
309	309	return
310	310	}
		@@ -354,6 +354,10 @@ func (s Server) streamBuild(w http.ResponseWriter, r http.Request, v buildView
354	354	path := v.Repo.Path()
355	355	msg, code := s.runControlStream(viewer, []string{"build", "log", path, n, "--follow"},
356	356	htmlStream{w: w, rc: rc}, r.Context().Done())
	357	if r.Context().Err() != nil {
	358	// The client left; nothing more to write.
	359	return
	360	}
357	361	if code == protocol.ExitDenied {
358	362	// The follow cap: the stored log once, and why it is not live.
359	363	log, _, _ := s.runControl(viewer, []string{"build", "log", path, n})
		@@ -367,7 +371,12 @@ func (s Server) streamBuild(w http.ResponseWriter, r http.Request, v buildView
367	371	fmt.Fprintf(w, `<p class="notice" role="status">build finished: %s</p>`, template.HTMLEscapeString(b.Status))
368	372	}
369	373	case code == protocol.ExitDenied:
	374	if viewer.ID == 0 {
	375	msg = "Too many signed-out viewers are watching live builds. This is the log so far; reload to try again, or sign in."
	376	}
370	377	fmt.Fprintf(w, `<p class="error" role="alert">%s</p>`, template.HTMLEscapeString(msg))
	378	case code == protocol.ExitFailure && msg != "":
	379	fmt.Fprintf(w, `<p class="notice" role="status">%s</p>`, template.HTMLEscapeString(msg))
371	380	}
372	381	io.WriteString(w, tail)
373	382	}
		@@ -380,7 +389,11 @@ type htmlStream struct {
380	389	}
381	390
382	391	func (h htmlStream) Write(p []byte) (int, error) {
383		template.HTMLEscape(h.w, p)
	392	var buf bytes.Buffer
	393	template.HTMLEscape(&buf, p)
	394	if _, err := h.w.Write(buf.Bytes()); err != nil {
	395	return 0, err
	396	}
384	397	if err := h.rc.Flush(); err != nil {
385	398	return 0, err
386	399	}

internal/httpd/compress.go +12 −2

		@@ -87,13 +87,23 @@ func (g *gzipWriter) Close() {
87	87	// Flush sends what the gzip stream holds, then flushes the connection, so
88	88	// a streamed page reaches the browser as it is written.
89	89	func (g *gzipWriter) Flush() {
	90	g.FlushError()
	91	}
	92
	93	// FlushError is Flush with the error a failed flush produces.
	94	// http.ResponseController.Flush prefers this over Flush when both are
	95	// implemented, so a write that fails partway through a stream is reported
	96	// instead of silently dropped.
	97	func (g *gzipWriter) FlushError() error {
90	98	if !g.decided {
91	99	g.decide(http.StatusOK)
92	100	}
93	101	if g.gz != nil {
94		g.gz.Flush()
	102	if err := g.gz.Flush(); err != nil {
	103	return err
	104	}
95	105	}
96		http.NewResponseController(g.ResponseWriter).Flush()
	106	return http.NewResponseController(g.ResponseWriter).Flush()
97	107	}
98	108
99	109	func (g *gzipWriter) Unwrap() http.ResponseWriter { return g.ResponseWriter }

Commit 3ac40e6851

Commit `3ac40e6851`