644155a37d · krz/gitbay

cmd/gitbayd/main.go +6

		@@ -325,6 +325,12 @@ func serveCmd() *cobra.Command {
325	325	ln.Close()
326	326	}
327	327	}
	328	// Follows run until a build ends; end them first so the drain
	329	// waits only for work that finishes.
	330	web.Stop()
	331	if sshSrv != nil {
	332	sshSrv.Stop()
	333	}
328	334	drain, cancel := context.WithTimeout(context.Background(), 30*time.Second)
329	335	defer cancel()
330	336	if err := hs.Shutdown(drain); err != nil {

e2e/buildfollow_test.go +25 −11

		@@ -54,15 +54,10 @@ func (s streamReader) waitFor(t testing.T, want string) string {
54	54	return s.buf.String()
55	55	}
56	56
57		// A running build is followed over ssh and on its page: output the runner
58		// sends arrives while the build runs, and both end with the outcome.
59		func TestBuildLogFollow(t *testing.T) {
60		t.Parallel()
61		inst := startInstance(t)
62		aliceKey := inst.newKey(t, "alice")
63		runnerKey := inst.newKey(t, "ci")
64		inst.admin(t, "admin", "user", "create", "alice", "--key", aliceKey+".pub")
65		inst.admin(t, "admin", "user", "create", "ci", "--key", runnerKey+".pub", "--admin")
	57	// queueBuild creates alice/app with one CI job and pushes it, which
	58	// queues build 1. No runner is attached, so it stays queued.
	59	func queueBuild(t testing.T, inst instance, aliceKey string) {
	60	t.Helper()
66	61	if _, _, code := inst.ssh(t, aliceKey, "", "repo", "create", "alice/app"); code != 0 {
67	62	t.Fatal("repo create failed")
68	63	}
		@@ -76,8 +71,12 @@ func TestBuildLogFollow(t *testing.T) {
76	71	mustGit(t, dir, env, "add", ".")
77	72	mustGit(t, dir, env, "commit", "-q", "-m", "ci")
78	73	mustGit(t, dir, env, "push", "-q", "origin", "main")
	74	}
79	75
80		// Claim build 1 by hand, so the test decides when output arrives.
	76	// claimBuild claims the oldest pending build with an admin key, as a
	77	// runner would, and returns its id for runner log and runner done.
	78	func claimBuild(t testing.T, inst instance, runnerKey string) string {
	79	t.Helper()
81	80	out, errOut, code := inst.ssh(t, runnerKey, "", "runner", "next", "--json")
82	81	if code != 0 {
83	82	t.Fatalf("runner next: %s", errOut)
		@@ -90,7 +89,22 @@ func TestBuildLogFollow(t *testing.T) {
90	89	if err := json.Unmarshal([]byte(out), &claim); err != nil \|\| claim.Data.ID == 0 {
91	90	t.Fatalf("runner next output %q: %v", out, err)
92	91	}
93		id := fmt.Sprint(claim.Data.ID)
	92	return fmt.Sprint(claim.Data.ID)
	93	}
	94
	95	// A running build is followed over ssh and on its page: output the runner
	96	// sends arrives while the build runs, and both end with the outcome.
	97	func TestBuildLogFollow(t *testing.T) {
	98	t.Parallel()
	99	inst := startInstance(t)
	100	aliceKey := inst.newKey(t, "alice")
	101	runnerKey := inst.newKey(t, "ci")
	102	inst.admin(t, "admin", "user", "create", "alice", "--key", aliceKey+".pub")
	103	inst.admin(t, "admin", "user", "create", "ci", "--key", runnerKey+".pub", "--admin")
	104	queueBuild(t, inst, aliceKey)
	105
	106	// Claim build 1 by hand, so the test decides when output arrives.
	107	id := claimBuild(t, inst, runnerKey)
94	108
95	109	cmd := inst.sshCmd(aliceKey, "build", "log", "alice/app", "1", "--follow")
96	110	stdout, err := cmd.StdoutPipe()

e2e/shutdown_test.go +74

		@@ -1,6 +1,7 @@
1	1	package e2e
2	2
3	3	import (
	4	"errors"
4	5	"fmt"
5	6	"net/http"
6	7	"os/exec"
		@@ -83,3 +84,76 @@ func TestShutdownClosesIdleConnections(t *testing.T) {
83	84	t.Fatalf("shutdown took %s with only an idle connection open", took)
84	85	}
85	86	}
	87
	88	// A deploy restarts the daemon while someone follows a build. The follows
	89	// end at once with a message saying so, rather than holding the drain for
	90	// its full 30 s and then being cut off mid-page.
	91	func TestShutdownEndsFollows(t *testing.T) {
	92	t.Parallel()
	93	inst := startInstance(t)
	94	aliceKey := inst.newKey(t, "alice")
	95	runnerKey := inst.newKey(t, "ci")
	96	inst.admin(t, "admin", "user", "create", "alice", "--key", aliceKey+".pub")
	97	inst.admin(t, "admin", "user", "create", "ci", "--key", runnerKey+".pub", "--admin")
	98	queueBuild(t, inst, aliceKey)
	99	// A line in the log is how the test knows both follows are streaming
	100	// before the signal; a follow still connecting is an idle connection,
	101	// which shutdown closes without a word.
	102	id := claimBuild(t, inst, runnerKey)
	103	if _, errOut, code := inst.ssh(t, runnerKey, "started\n", "runner", "log", id); code != 0 {
	104	t.Fatalf("runner log: %s", errOut)
	105	}
	106
	107	cmd := inst.sshCmd(aliceKey, "build", "log", "alice/app", "1", "--follow")
	108	stdout, err := cmd.StdoutPipe()
	109	if err != nil {
	110	t.Fatal(err)
	111	}
	112	var stderr strings.Builder
	113	cmd.Stderr = &stderr
	114	if err := cmd.Start(); err != nil {
	115	t.Fatal(err)
	116	}
	117	defer func() {
	118	if cmd.ProcessState == nil {
	119	cmd.Process.Kill()
	120	cmd.Wait()
	121	}
	122	}()
	123
	124	page, err := http.Get(inst.base() + "/alice/app/builds/1")
	125	if err != nil {
	126	t.Fatal(err)
	127	}
	128	defer page.Body.Close()
	129	web := newStreamReader(page.Body)
	130	web.waitFor(t, "started")
	131	newStreamReader(stdout).waitFor(t, "started")
	132
	133	start := time.Now()
	134	if err := inst.proc.Process.Signal(syscall.SIGTERM); err != nil {
	135	t.Fatal(err)
	136	}
	137	web.waitFor(t, "gitbay is restarting; reload in a moment")
	138	web.waitFor(t, "</html>")
	139	var exit *exec.ExitError
	140	if err := cmd.Wait(); !errors.As(err, &exit) \|\| exit.ExitCode() != 1 {
	141	t.Fatalf("follow ended with %v, want exit 1\n%s", err, stderr.String())
	142	}
	143	if !strings.Contains(stderr.String(), "gitbay is restarting") {
	144	t.Errorf("follow stderr %q", stderr.String())
	145	}
	146	done := make(chan error, 1)
	147	go func() { done <- inst.proc.Wait() }()
	148	select {
	149	case err := <-done:
	150	if err != nil {
	151	t.Fatalf("daemon did not exit cleanly: %v", err)
	152	}
	153	case <-time.After(20 * time.Second):
	154	t.Fatal("daemon still running 20s after SIGTERM")
	155	}
	156	if took := time.Since(start); took > 5*time.Second {
	157	t.Fatalf("shutdown took %s with two follows open", took)
	158	}
	159	}

internal/httpd/api.go +1 −1

		@@ -73,7 +73,7 @@ func (s Server) apiCmd(w http.ResponseWriter, r http.Request) {
73	73	JSON: true,
74	74	ViaAPI: true,
75	75	ReadOnly: scope == "read",
76		Done: r.Context().Done(),
	76	Done: s.until(r),
77	77	}
78	78	code := control.Dispatch(ctx, req.Argv)
79	79

internal/httpd/apiread.go +1 −1

		@@ -62,7 +62,7 @@ func (s Server) apiRead(w http.ResponseWriter, r http.Request) {
62	62	JSON: true,
63	63	ViaAPI: true,
64	64	ReadOnly: true,
65		Done: r.Context().Done(),
	65	Done: s.until(r),
66	66	}
67	67	code := control.Dispatch(ctx, argv)
68	68

internal/httpd/builds.go +6 −1

		@@ -353,11 +353,16 @@ func (s Server) streamBuild(w http.ResponseWriter, r http.Request, v buildView
353	353
354	354	path := v.Repo.Path()
355	355	msg, code := s.runControlStream(viewer, []string{"build", "log", path, n, "--follow"},
356		htmlStream{w: w, rc: rc}, r.Context().Done())
	356	htmlStream{w: w, rc: rc}, s.until(r))
357	357	if r.Context().Err() != nil {
358	358	// The client left; nothing more to write.
359	359	return
360	360	}
	361	if code != protocol.ExitOK && s.stopped() {
	362	io.WriteString(w, `</pre><p class="notice" role="status">gitbay is restarting; reload in a moment to pick the log up again.</p>`)
	363	io.WriteString(w, tail)
	364	return
	365	}
361	366	if code == protocol.ExitDenied {
362	367	// The follow cap: the stored log once, and why it is not live.
363	368	log, _, _ := s.runControl(viewer, []string{"build", "log", path, n})

internal/httpd/smart.go +35 −1

		@@ -14,6 +14,7 @@ import (
14	14	"os"
15	15	"os/exec"
16	16	"strings"
	17	"sync"
17	18
18	19	"gitbay.org/gitbay/internal/config"
19	20	"gitbay.org/gitbay/internal/control"
		@@ -26,11 +27,44 @@ type Server struct {
26	27	st *store.Store
27	28	apiLimit *apiLimiter
28	29	proxies []*net.IPNet // http.trusted_proxies, parsed once
	30	stopping chan struct{} // closed by Stop
	31	stopOnce sync.Once
29	32	}
30	33
31	34	func New(cfg config.Config, st store.Store) Server {
32	35	proxies, _ := cfg.HTTP.TrustedProxyNets() // validated at config load
33		return &Server{cfg: cfg, st: st, apiLimit: newAPILimiter(cfg.Limits.APIRate), proxies: proxies}
	36	return &Server{cfg: cfg, st: st, apiLimit: newAPILimiter(cfg.Limits.APIRate), proxies: proxies,
	37	stopping: make(chan struct{})}
	38	}
	39
	40	// Stop ends the requests running a command that lasts until something
	41	// happens (build log --follow), so a shutdown drain waits only for work
	42	// that finishes. Other requests, git transport included, run on.
	43	func (s *Server) Stop() {
	44	s.stopOnce.Do(func() { close(s.stopping) })
	45	}
	46
	47	// until is closed when the request ends or the server stops, whichever
	48	// comes first: the Done a following command runs under.
	49	func (s Server) until(r http.Request) <-chan struct{} {
	50	done := make(chan struct{})
	51	go func() {
	52	select {
	53	case <-r.Context().Done():
	54	case <-s.stopping:
	55	}
	56	close(done)
	57	}()
	58	return done
	59	}
	60
	61	func (s *Server) stopped() bool {
	62	select {
	63	case <-s.stopping:
	64	return true
	65	default:
	66	return false
	67	}
34	68	}
35	69
36	70	// receivePackRefusal exists only to fail legibly if a client POSTs without

Commit 644155a37d

Commit `644155a37d`