!463 · krz/gitbay

.gitbay/wiki/CI.org +4 −1

		@@ -47,7 +47,10 @@ queued build. The build page does the same without JavaScript while a
47	47	build is queued or running; =?follow=0= renders it once. An account
48	48	holds at most eight follows open, and signed-out viewers share one
49	49	account's eight. Over the JSON API the command answers when the build
50		ends, with the whole log.
	50	ends, with the whole log. Read access is checked again every two
	51	seconds: a follower who loses it is told the repository is not found.
	52	A restart ends every open follow with a message saying so, rather
	53	than holding the drain; follow again once the daemon is back.
51	54
52	55	* The table
53	56

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 {

cmd/gitbayd/system.go +1 −1

		@@ -93,7 +93,7 @@ func shellCmd() *cobra.Command {
93	93	fmt.Fprintf(os.Stderr, "gitbay control plane: interactive shells are not available.\nTry: ssh <host> help\n")
94	94	os.Exit(protocol.ExitUsage)
95	95	}
96		code := sshd.Exec(cfg, st, user, key.Scope, key.Fingerprint, cmdline, os.Stdin, os.Stdout, os.Stderr, nil)
	96	code := sshd.Exec(cfg, st, user, key.Scope, key.Fingerprint, cmdline, os.Stdin, os.Stdout, os.Stderr, nil, nil)
97	97	st.Close()
98	98	os.Exit(code)
99	99	return 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, `<p class="notice" role="status">gitbay is restarting`)
	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/control/build.go +2 −2

		@@ -200,12 +200,12 @@ func runBuildLog(c *Ctx, args []string) int {
200	200	if err != nil {
201	201	return c.fail(protocol.ExitUsage, "%v", err)
202	202	}
203		_, b, code := buildRef(c, f.Pos)
	203	repo, b, code := buildRef(c, f.Pos)
204	204	if code >= 0 {
205	205	return code
206	206	}
207	207	if f.Has("--follow") {
208		return followBuildLog(c, b)
	208	return followBuildLog(c, repo, b)
209	209	}
210	210	log, err := c.Store.BuildLog(b.ID)
211	211	if err != nil {

internal/control/buildfollow.go +74 −3

		@@ -1,10 +1,12 @@
1	1	package control
2	2
3	3	import (
	4	"errors"
4	5	"fmt"
5	6	"sync"
6	7	"time"
7	8
	9	"gitbay.org/gitbay/internal/policy"
8	10	"gitbay.org/gitbay/internal/protocol"
9	11	"gitbay.org/gitbay/internal/store"
10	12	)
		@@ -28,6 +30,11 @@ var (
28	30	// reaper's deadline, so a follow needs its own limit for the queued
29	31	// case or it never ends.
30	32	followQueued = 10 * time.Minute
	33	// followCoalesce is the pause between a wake and the read it causes.
	34	// A runner appends a chunk per read of its output, often a line, and
	35	// each read loads the whole stored log; one read after a short pause
	36	// takes a burst of appends together.
	37	followCoalesce = 200 * time.Millisecond
31	38	)
32	39
33	40	var (
		@@ -53,10 +60,56 @@ func dropFollow(uid int64) {
53	60	}
54	61	}
55	62
	63	// mayStillRead reports whether the follower can still read the build's
	64	// repository. It looks the repository up by id, so a rename mid-follow
	65	// does not end the follow, and reloads the account, so disabling it
	66	// does.
	67	func mayStillRead(c *Ctx, repoID int64) (bool, error) {
	68	repo, err := c.Store.RepoByID(repoID)
	69	if errors.Is(err, store.ErrNotFound) {
	70	return false, nil
	71	}
	72	if err != nil {
	73	return false, err
	74	}
	75	u := c.User
	76	if u.ID != 0 {
	77	u, err = c.Store.UserByID(u.ID)
	78	if errors.Is(err, store.ErrNotFound) {
	79	return false, nil
	80	}
	81	if err != nil {
	82	return false, err
	83	}
	84	if u.Disabled {
	85	return false, nil
	86	}
	87	}
	88	grant, err := c.Store.AccessRole(repo.ID, u.ID)
	89	if err != nil {
	90	return false, err
	91	}
	92	return policy.CanRead(u, repo, grant), nil
	93	}
	94
	95	// ended is what a follow returns when its Done closes. A restart says
	96	// so, whatever the surface, so the reader knows to follow again; a
	97	// reader who left hears nothing.
	98	func ended(c *Ctx) int {
	99	select {
	100	case <-c.Stopping:
	101	fmt.Fprintln(c.Stderr, "gitbay is restarting; follow the build again in a moment")
	102	default:
	103	}
	104	return protocol.ExitFailure
	105	}
	106
56	107	// followBuildLog writes the build's log as it grows and returns once the
57	108	// build has an outcome and its last bytes are written. The outcome goes
58		// to stderr, so stdout is the log byte for byte.
59		func followBuildLog(c *Ctx, b store.Build) int {
	109	// to stderr, so stdout is the log byte for byte. Read access is checked
	110	// again every followPoll: a repository made private, or a grant revoked,
	111	// ends the follow with the answer a new request would get.
	112	func followBuildLog(c *Ctx, repo store.Repo, b store.Build) int {
60	113	if !takeFollow(c.User.ID) {
61	114	return c.fail(protocol.ExitDenied, "%d follows are already open for this account; close one and retry", maxFollows)
62	115	}
		@@ -65,7 +118,18 @@ func followBuildLog(c *Ctx, b store.Build) int {
65	118	var off int64
66	119	var settleBy time.Time
67	120	var queuedSince time.Time
	121	checked := time.Now()
68	122	for {
	123	if time.Since(checked) >= followPoll {
	124	ok, err := mayStillRead(c, b.RepoID)
	125	if err != nil {
	126	return c.fail(protocol.ExitFailure, "%v", err)
	127	}
	128	if !ok {
	129	return c.fail(protocol.ExitNotFound, "repository %s not found", repo.Path())
	130	}
	131	checked = time.Now()
	132	}
69	133	wake := c.Store.BuildLogWait(b.ID)
70	134	status, chunk, err := c.Store.BuildLogFrom(b.ID, off)
71	135	if err != nil {
		@@ -107,10 +171,17 @@ func followBuildLog(c *Ctx, b store.Build) int {
107	171	t := time.NewTimer(wait)
108	172	select {
109	173	case <-wake:
	174	t.Reset(followCoalesce)
	175	select {
	176	case <-t.C:
	177	case <-c.Done:
	178	t.Stop()
	179	return ended(c)
	180	}
110	181	case <-t.C:
111	182	case <-c.Done:
112	183	t.Stop()
113		return protocol.ExitFailure
	184	return ended(c)
114	185	}
115	186	t.Stop()
116	187	}

internal/control/buildfollow_test.go +103 −13

		@@ -33,6 +33,13 @@ func (b *syncBuffer) String() string {
33	33	// follow starts build log --follow on build 1 of repo and returns the
34	34	// buffers and a channel carrying the exit code.
35	35	func follow(t testing.T, st store.Store, uid int64, repo store.Repo, done <-chan struct{}) (syncBuffer, syncBuffer, chan int) {
	36	t.Helper()
	37	return followStopping(t, st, uid, repo, done, nil)
	38	}
	39
	40	// followStopping is follow on a surface that is being restarted when
	41	// stopping closes.
	42	func followStopping(t testing.T, st store.Store, uid int64, repo store.Repo, done, stopping <-chan struct{}) (syncBuffer, syncBuffer, chan int) {
36	43	t.Helper()
37	44	u, err := st.UserByID(uid)
38	45	if err != nil {
		@@ -40,12 +47,26 @@ func follow(t testing.T, st store.Store, uid int64, repo store.Repo, done <-ch
40	47	}
41	48	var out, errOut syncBuffer
42	49	c := &Ctx{User: u, Scope: "full", Store: st, Stdin: strings.NewReader(""),
43		Stdout: &out, Stderr: &errOut, Done: done}
	50	Stdout: &out, Stderr: &errOut, Done: done, Stopping: stopping}
44	51	res := make(chan int, 1)
45	52	go func() { res <- Dispatch(c, []string{"build", "log", repo.Path(), "1", "--follow"}) }()
46	53	return &out, &errOut, res
47	54	}
48	55
	56	// waitOutput waits until the follower has written want, which is how a
	57	// test knows the follow is past its first read.
	58	func waitOutput(t testing.T, out syncBuffer, want string) {
	59	t.Helper()
	60	deadline := time.After(2 * time.Second)
	61	for !strings.Contains(out.String(), want) {
	62	select {
	63	case <-deadline:
	64	t.Fatalf("follow never wrote %q", want)
	65	case <-time.After(10 * time.Millisecond):
	66	}
	67	}
	68	}
	69
49	70	func waitExit(t *testing.T, res chan int) int {
50	71	t.Helper()
51	72	select {
		@@ -58,9 +79,9 @@ func waitExit(t *testing.T, res chan int) int {
58	79	}
59	80
60	81	func shortFollowTimers(t *testing.T) {
61		settle, poll, queued := followSettle, followPoll, followQueued
62		followSettle, followPoll = 200time.Millisecond, 50time.Millisecond
63		t.Cleanup(func() { followSettle, followPoll, followQueued = settle, poll, queued })
	82	settle, poll, queued, coalesce := followSettle, followPoll, followQueued, followCoalesce
	83	followSettle, followPoll, followCoalesce = 200time.Millisecond, 50time.Millisecond, 10*time.Millisecond
	84	t.Cleanup(func() { followSettle, followPoll, followQueued, followCoalesce = settle, poll, queued, coalesce })
64	85	}
65	86
66	87	// The follow prints the stored log, then what arrives, and ends with the
		@@ -132,20 +153,62 @@ func TestBuildLogFollowDone(t *testing.T) {
132	153	}
133	154	st.AppendBuildLog(id, []byte("step one\n"))
134	155	done := make(chan struct{})
135		out, _, res := follow(t, st, uid, repo, done)
	156	out, errOut, res := follow(t, st, uid, repo, done)
136	157
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		}
	158	waitOutput(t, out, "step one")
145	159	close(done)
146	160	if code := waitExit(t, res); code != protocol.ExitFailure {
147	161	t.Fatalf("exit %d, want %d", code, protocol.ExitFailure)
148	162	}
	163	if errOut.String() != "" {
	164	t.Errorf("a follow whose reader left wrote %q", errOut)
	165	}
	166	}
	167
	168	// A restart ends a follow and says so, so the reader knows to follow
	169	// again.
	170	func TestBuildLogFollowRestart(t *testing.T) {
	171	shortFollowTimers(t)
	172	st, repo, uid := newQueueTestRepo(t)
	173	id, err := st.CreateBuild(repo.ID, "unit", "abc", "main", `["true"]`, "", "", true)
	174	if err != nil {
	175	t.Fatal(err)
	176	}
	177	st.AppendBuildLog(id, []byte("step one\n"))
	178	stopping := make(chan struct{})
	179	out, errOut, res := followStopping(t, st, uid, repo, stopping, stopping)
	180	waitOutput(t, out, "step one")
	181	close(stopping)
	182	if code := waitExit(t, res); code != protocol.ExitFailure {
	183	t.Fatalf("exit %d, want %d", code, protocol.ExitFailure)
	184	}
	185	if got := strings.TrimSpace(errOut.String()); got != "gitbay is restarting; follow the build again in a moment" {
	186	t.Errorf("stderr %q", got)
	187	}
	188	}
	189
	190	// A follower whose account is disabled mid-follow is ended, even on a
	191	// public repository.
	192	func TestBuildLogFollowDisabled(t *testing.T) {
	193	shortFollowTimers(t)
	194	st, repo, _ := newQueueTestRepo(t)
	195	bob, err := st.CreateUser("bob", false)
	196	if err != nil {
	197	t.Fatal(err)
	198	}
	199	id, err := st.CreateBuild(repo.ID, "unit", "abc", "main", `["true"]`, "", "", true)
	200	if err != nil {
	201	t.Fatal(err)
	202	}
	203	st.AppendBuildLog(id, []byte("step one\n"))
	204	out, errOut, res := follow(t, st, bob, repo, nil)
	205	waitOutput(t, out, "step one")
	206	if err := st.SetUserDisabled(bob, true); err != nil {
	207	t.Fatal(err)
	208	}
	209	if code := waitExit(t, res); code != protocol.ExitNotFound {
	210	t.Fatalf("exit %d, want %d: %s", code, protocol.ExitNotFound, errOut)
	211	}
149	212	}
150	213
151	214	// A build that stays pending ends its own follow: nothing reaps a queued
		@@ -166,6 +229,33 @@ func TestBuildLogFollowQueued(t *testing.T) {
166	229	}
167	230	}
168	231
	232	// A follower who loses read access mid-follow is ended with the answer
	233	// a new request would get: the repository is not found.
	234	func TestBuildLogFollowLosesAccess(t *testing.T) {
	235	shortFollowTimers(t)
	236	st, repo, _ := newQueueTestRepo(t)
	237	bob, err := st.CreateUser("bob", false)
	238	if err != nil {
	239	t.Fatal(err)
	240	}
	241	id, err := st.CreateBuild(repo.ID, "unit", "abc", "main", `["true"]`, "", "", true)
	242	if err != nil {
	243	t.Fatal(err)
	244	}
	245	st.AppendBuildLog(id, []byte("step one\n"))
	246	out, errOut, res := follow(t, st, bob, repo, nil)
	247	waitOutput(t, out, "step one")
	248	if err := st.SetRepoVisibility(repo.ID, "private"); err != nil {
	249	t.Fatal(err)
	250	}
	251	if code := waitExit(t, res); code != protocol.ExitNotFound {
	252	t.Fatalf("exit %d, want %d: %s", code, protocol.ExitNotFound, errOut)
	253	}
	254	if !strings.Contains(errOut.String(), "repository "+repo.Path()+" not found") {
	255	t.Errorf("stderr %q", errOut)
	256	}
	257	}
	258
169	259	// An account holding maxFollows is refused another.
170	260	func TestBuildLogFollowCap(t *testing.T) {
171	261	st, repo, uid := newQueueTestRepo(t)

internal/control/control.go +4

		@@ -44,6 +44,10 @@ type Ctx struct {
44	44	// more: the SSH channel closed or the HTTP request ended. A command
45	45	// that runs until something happens (build log --follow) stops on it.
46	46	Done <-chan struct{}
	47	// Stopping, when the surface has one, closes when the daemon is
	48	// restarting. It closes Done too; a command that ends on Done checks
	49	// it to say why.
	50	Stopping <-chan struct{}
47	51	}
48	52
49	53	// usage reports a bad invocation with the command's registered usage,

internal/httpd/api.go +2 −1

		@@ -73,7 +73,8 @@ 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	Stopping: s.stopping,
77	78	}
78	79	code := control.Dispatch(ctx, req.Argv)
79	80

internal/httpd/apiread.go +2 −1

		@@ -62,7 +62,8 @@ 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	Stopping: s.stopping,
66	67	}
67	68	code := control.Dispatch(ctx, argv)
68	69

internal/httpd/builds.go +1 −1

		@@ -353,7 +353,7 @@ 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

internal/httpd/control.go +11 −10

		@@ -57,16 +57,17 @@ func (s *Server) runControlCode(u store.User, argv []string) (out string, msg st
57	57	func (s *Server) runControlStream(u store.User, argv []string, out io.Writer, done <-chan struct{}) (msg string, code int) {
58	58	var stderr bytes.Buffer
59	59	ctx := &control.Ctx{
60		User: u,
61		Source: "web",
62		Scope: "full",
63		Store: s.st,
64		Cfg: s.cfg,
65		Stdin: strings.NewReader(""),
66		Stdout: out,
67		Stderr: &stderr,
68		ViaAPI: true,
69		Done: done,
	60	User: u,
	61	Source: "web",
	62	Scope: "full",
	63	Store: s.st,
	64	Cfg: s.cfg,
	65	Stdin: strings.NewReader(""),
	66	Stdout: out,
	67	Stderr: &stderr,
	68	ViaAPI: true,
	69	Done: done,
	70	Stopping: s.stopping,
70	71	}
71	72	code = control.Dispatch(ctx, argv)
72	73	return strings.TrimSpace(stderr.String()), code

internal/httpd/smart.go +27 −2

		@@ -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"
		@@ -25,12 +26,36 @@ type Server struct {
25	26	cfg config.Config
26	27	st *store.Store
27	28	apiLimit *apiLimiter
28		proxies []*net.IPNet // http.trusted_proxies, parsed once
	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
34	59	}
35	60
36	61	// receivePackRefusal exists only to fail legibly if a client POSTs without

internal/sshd/sshd.go +34 −14

		@@ -39,6 +39,8 @@ type Server struct {
39	39	sessions sync.WaitGroup // accepted connections still being served
40	40	mu sync.Mutex
41	41	conns map[*conn]struct{}
	42	stopping chan struct{} // closed by Stop
	43	stopOnce sync.Once
42	44	}
43	45
44	46	// conn is one accepted connection and how many sessions it is running.
		@@ -51,7 +53,7 @@ type conn struct {
51	53	}
52	54
53	55	func New(cfg config.Config, st store.Store) (Server, error) {
54		s := &Server{cfg: cfg, st: st, authLimiter: newRateLimiter(cfg.Limits.SSHAuthRate, time.Minute), conns: map[*conn]struct{}{}}
	56	s := &Server{cfg: cfg, st: st, authLimiter: newRateLimiter(cfg.Limits.SSHAuthRate, time.Minute), conns: map[*conn]struct{}{}, stopping: make(chan struct{})}
55	57
56	58	sc := &ssh.ServerConfig{
57	59	PublicKeyCallback: s.authenticate,
		@@ -177,10 +179,18 @@ func (s *Server) Serve(ln net.Listener) error {
177	179	}
178	180	}
179	181
	182	// Stop ends the commands that run until something happens (build log
	183	// --follow), so a shutdown drain waits only for work that finishes. It
	184	// does not close connections; Shutdown does.
	185	func (s *Server) Stop() {
	186	s.stopOnce.Do(func() { close(s.stopping) })
	187	}
	188
180	189	// Shutdown closes every idle connection, then waits for the ones with a
181	190	// session running, or for ctx. The caller closes the listener first; a
182	191	// push in flight completes rather than being cut mid-pack.
183	192	func (s *Server) Shutdown(ctx context.Context) error {
	193	s.Stop()
184	194	s.mu.Lock()
185	195	for c := range s.conns {
186	196	if c.active.Load() == 0 {
		@@ -240,12 +250,21 @@ func (s Server) handleSession(sconn ssh.ServerConn, ch ssh.Channel, reqs <-cha
240	250	req.Reply(true, nil)
241	251	// x/crypto closes reqs when the client closes the channel. That
242	252	// is how a follow learns nobody is reading: the CLI's shared
243		// connection outlives a Ctrl-C, the channel does not.
244		done := make(chan struct{})
	253	// connection outlives a Ctrl-C, the channel does not. Stop
	254	// ends it too, for a restart.
	255	closed := make(chan struct{})
245	256	go func() {
246	257	for r := range reqs {
247	258	r.Reply(false, nil)
248	259	}
	260	close(closed)
	261	}()
	262	done := make(chan struct{})
	263	go func() {
	264	select {
	265	case <-closed:
	266	case <-s.stopping:
	267	}
249	268	close(done)
250	269	}()
251	270	code := s.runExec(sconn, ch, payload.Command, done)
		@@ -283,7 +302,7 @@ func (s Server) runExec(sconn ssh.ServerConn, ch ssh.Channel, cmdline string,
283	302	return protocol.ExitDenied
284	303	}
285	304	_ = s.st.TouchSSHKey(keyID)
286		return Exec(s.cfg, s.st, user, ext["scope"], ext["key-fp"], cmdline, ch, ch, ch.Stderr(), done)
	305	return Exec(s.cfg, s.st, user, ext["scope"], ext["key-fp"], cmdline, ch, ch, ch.Stderr(), done, s.stopping)
287	306	}
288	307
289	308	// runAnonymous handles a session from an unregistered key: the register
		@@ -314,7 +333,7 @@ func (s *Server) runAnonymous(ch ssh.Channel, keyB64, cmdline string) int {
314	333	// single dispatch path shared by the embedded listener and the system-sshd
315	334	// forced command (gitbayd shell).
316	335	func Exec(cfg config.Config, st *store.Store, user store.User, scope, source, cmdline string,
317		stdin io.Reader, stdout, stderr io.Writer, done <-chan struct{}) int {
	336	stdin io.Reader, stdout, stderr io.Writer, done, stopping <-chan struct{}) int {
318	337	if user.Disabled {
319	338	fmt.Fprintln(stderr, "this account is disabled; contact the instance admin")
320	339	return protocol.ExitDenied
		@@ -343,15 +362,16 @@ func Exec(cfg config.Config, st *store.Store, user store.User, scope, source, cm
343	362	}
344	363	}
345	364	ctx := &control.Ctx{
346		User: user,
347		Scope: scope,
348		Source: source,
349		Store: st,
350		Cfg: cfg,
351		Stdin: stdin,
352		Stdout: stdout,
353		Stderr: stderr,
354		Done: done,
	365	User: user,
	366	Scope: scope,
	367	Source: source,
	368	Store: st,
	369	Cfg: cfg,
	370	Stdin: stdin,
	371	Stdout: stdout,
	372	Stderr: stderr,
	373	Done: done,
	374	Stopping: stopping,
355	375	}
356	376	return control.Dispatch(ctx, argv)
357	377	}

internal/sshd/sshd_test.go added +199

		@@ -0,0 +1,199 @@
	1	package sshd
	2
	3	import (
	4	"bufio"
	5	"bytes"
	6	"crypto/ed25519"
	7	"crypto/rand"
	8	"errors"
	9	"net"
	10	"path/filepath"
	11	"strings"
	12	"testing"
	13	"time"
	14
	15	"golang.org/x/crypto/ssh"
	16
	17	"gitbay.org/gitbay/internal/config"
	18	"gitbay.org/gitbay/internal/store"
	19	)
	20
	21	// followServer starts an embedded server holding alice, her public repo
	22	// alice/app and a queued build 1 whose log has one line, and returns it
	23	// with a client connected as alice.
	24	func followServer(t testing.T) (Server, *ssh.Client) {
	25	t.Helper()
	26	root := t.TempDir()
	27	st, err := store.Open(filepath.Join(root, "gitbay.db"))
	28	if err != nil {
	29	t.Fatal(err)
	30	}
	31	t.Cleanup(func() { st.Close() })
	32	if err := st.MigrateUp(); err != nil {
	33	t.Fatal(err)
	34	}
	35	uid, err := st.CreateUser("alice", false)
	36	if err != nil {
	37	t.Fatal(err)
	38	}
	39	repoID, err := st.CreateRepo("user", uid, "app", "public")
	40	if err != nil {
	41	t.Fatal(err)
	42	}
	43	id, err := st.CreateBuild(repoID, "unit", "abc", "main", `["true"]`, "", "", true)
	44	if err != nil {
	45	t.Fatal(err)
	46	}
	47	if err := st.AppendBuildLog(id, []byte("queued\n")); err != nil {
	48	t.Fatal(err)
	49	}
	50	_, priv, err := ed25519.GenerateKey(rand.Reader)
	51	if err != nil {
	52	t.Fatal(err)
	53	}
	54	signer, err := ssh.NewSignerFromKey(priv)
	55	if err != nil {
	56	t.Fatal(err)
	57	}
	58	pub := signer.PublicKey()
	59	if err := st.AddSSHKey(uid, ssh.FingerprintSHA256(pub), pub.Type(), pub.Marshal(), "full", "test"); err != nil {
	60	t.Fatal(err)
	61	}
	62
	63	cfg := config.Default()
	64	cfg.Server.Root = root
	65	srv, err := New(cfg, st)
	66	if err != nil {
	67	t.Fatal(err)
	68	}
	69	ln, err := net.Listen("tcp", "127.0.0.1:0")
	70	if err != nil {
	71	t.Fatal(err)
	72	}
	73	go srv.Serve(ln)
	74	t.Cleanup(func() { ln.Close() })
	75
	76	client, err := ssh.Dial("tcp", ln.Addr().String(), &ssh.ClientConfig{
	77	User: "git",
	78	Auth: []ssh.AuthMethod{ssh.PublicKeys(signer)},
	79	HostKeyCallback: ssh.InsecureIgnoreHostKey(),
	80	Timeout: 5 * time.Second,
	81	})
	82	if err != nil {
	83	t.Fatal(err)
	84	}
	85	t.Cleanup(func() { client.Close() })
	86	return srv, client
	87	}
	88
	89	// startFollow runs build log --follow on a new session and returns once
	90	// the stored line has arrived, so the follow is past its first read.
	91	func startFollow(t testing.T, client ssh.Client, stderr bytes.Buffer) ssh.Session {
	92	t.Helper()
	93	sess, err := client.NewSession()
	94	if err != nil {
	95	t.Fatal(err)
	96	}
	97	t.Cleanup(func() { sess.Close() })
	98	sess.Stderr = stderr
	99	out, err := sess.StdoutPipe()
	100	if err != nil {
	101	t.Fatal(err)
	102	}
	103	if err := sess.Start("build log alice/app 1 --follow"); err != nil {
	104	t.Fatal(err)
	105	}
	106	line := make(chan string, 1)
	107	go func() {
	108	l, _ := bufio.NewReader(out).ReadString('\n')
	109	line <- l
	110	}()
	111	select {
	112	case l := <-line:
	113	if l != "queued\n" {
	114	t.Fatalf("first line %q", l)
	115	}
	116	case <-time.After(5 * time.Second):
	117	t.Fatal("the stored log never arrived")
	118	}
	119	return sess
	120	}
	121
	122	func activeSessions(s *Server) int32 {
	123	s.mu.Lock()
	124	defer s.mu.Unlock()
	125	var n int32
	126	for c := range s.conns {
	127	n += c.active.Load()
	128	}
	129	return n
	130	}
	131
	132	// Closing the session channel ends a follow of a queued build while the
	133	// connection stays up, as a Ctrl-C does over the CLI's shared connection.
	134	// Nothing else would end it for ten minutes.
	135	func TestFollowEndsWhenChannelCloses(t *testing.T) {
	136	srv, client := followServer(t)
	137	var stderr bytes.Buffer
	138	sess := startFollow(t, client, &stderr)
	139	if n := activeSessions(srv); n != 1 {
	140	t.Fatalf("%d sessions active while following, want 1", n)
	141	}
	142	sess.Close()
	143
	144	deadline := time.Now().Add(5 * time.Second)
	145	for activeSessions(srv) != 0 {
	146	if time.Now().After(deadline) {
	147	t.Fatal("the follow outlived its channel")
	148	}
	149	time.Sleep(20 * time.Millisecond)
	150	}
	151	if _, err := client.NewSession(); err != nil {
	152	t.Fatalf("the connection did not survive the channel: %v", err)
	153	}
	154	}
	155
	156	// A command that fails on its own while the server is stopping says
	157	// nothing about a restart: only a follow that Stop ended does.
	158	func TestStopLeavesOtherFailuresAlone(t *testing.T) {
	159	srv, client := followServer(t)
	160	srv.Stop()
	161	sess, err := client.NewSession()
	162	if err != nil {
	163	t.Fatal(err)
	164	}
	165	defer sess.Close()
	166	var stderr bytes.Buffer
	167	sess.Stderr = &stderr
	168	var exit *ssh.ExitError
	169	if err := sess.Run("repo show nosuch/repo"); !errors.As(err, &exit) \|\| exit.ExitStatus() != 3 {
	170	t.Fatalf("repo show of a missing repository: %v, want exit 3", err)
	171	}
	172	if strings.Contains(stderr.String(), "restarting") {
	173	t.Errorf("stderr %q", stderr.String())
	174	}
	175	}
	176
	177	// Stop ends a follow with exit 1 and says why, without closing the
	178	// connection.
	179	func TestStopEndsFollow(t *testing.T) {
	180	srv, client := followServer(t)
	181	var stderr bytes.Buffer
	182	sess := startFollow(t, client, &stderr)
	183	srv.Stop()
	184
	185	waited := make(chan error, 1)
	186	go func() { waited <- sess.Wait() }()
	187	select {
	188	case err := <-waited:
	189	var exit *ssh.ExitError
	190	if !errors.As(err, &exit) \|\| exit.ExitStatus() != 1 {
	191	t.Fatalf("follow ended with %v, want exit 1", err)
	192	}
	193	case <-time.After(5 * time.Second):
	194	t.Fatal("Stop did not end the follow")
	195	}
	196	if !strings.Contains(stderr.String(), "gitbay is restarting") {
	197	t.Errorf("stderr %q", stderr.String())
	198	}
	199	}

build log --follow: end on restart, re-check access, coalesce reads !463