krz/gitbay

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

merged merged by cmc on 2026-09-23 07:12 UTC · krz/gitbay:follow-followups into main

16 files changed, +569 −60

Layout: unified · split

.gitbay/wiki/CI.org +4 −1
@@ -47,7 +47,10 @@ queued build. The build page does the same without JavaScript while a
47build is queued or running; =?follow=0= renders it once. An account 47build is queued or running; =?follow=0= renders it once. An account
48holds at most eight follows open, and signed-out viewers share one 48holds at most eight follows open, and signed-out viewers share one
49account's eight. Over the JSON API the command answers when the build 49account's eight. Over the JSON API the command answers when the build
50ends, with the whole log. 50ends, with the whole log. Read access is checked again every two
51seconds: a follower who loses it is told the repository is not found.
52A restart ends every open follow with a message saying so, rather
53than holding the drain; follow again once the daemon is back.
51 54
52* The table 55* The table
53 56
cmd/gitbayd/main.go +6
@@ -325,6 +325,12 @@ func serveCmd() *cobra.Command {
325 ln.Close() 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 drain, cancel := context.WithTimeout(context.Background(), 30*time.Second) 334 drain, cancel := context.WithTimeout(context.Background(), 30*time.Second)
329 defer cancel() 335 defer cancel()
330 if err := hs.Shutdown(drain); err != nil { 336 if err := hs.Shutdown(drain); err != nil {
cmd/gitbayd/system.go +1 −1
@@ -93,7 +93,7 @@ func shellCmd() *cobra.Command {
93 fmt.Fprintf(os.Stderr, "gitbay control plane: interactive shells are not available.\nTry: ssh <host> help\n") 93 fmt.Fprintf(os.Stderr, "gitbay control plane: interactive shells are not available.\nTry: ssh <host> help\n")
94 os.Exit(protocol.ExitUsage) 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 st.Close() 97 st.Close()
98 os.Exit(code) 98 os.Exit(code)
99 return nil 99 return nil
e2e/buildfollow_test.go +25 −11
@@ -54,15 +54,10 @@ func (s *streamReader) waitFor(t *testing.T, want string) string {
54 return s.buf.String() 54 return s.buf.String()
55} 55}
56 56
57// A running build is followed over ssh and on its page: output the runner 57// queueBuild creates alice/app with one CI job and pushes it, which
58// sends arrives while the build runs, and both end with the outcome. 58// queues build 1. No runner is attached, so it stays queued.
59func TestBuildLogFollow(t *testing.T) { 59func queueBuild(t *testing.T, inst *instance, aliceKey string) {
60 t.Parallel() 60 t.Helper()
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")
66 if _, _, code := inst.ssh(t, aliceKey, "", "repo", "create", "alice/app"); code != 0 { 61 if _, _, code := inst.ssh(t, aliceKey, "", "repo", "create", "alice/app"); code != 0 {
67 t.Fatal("repo create failed") 62 t.Fatal("repo create failed")
68 } 63 }
@@ -76,8 +71,12 @@ func TestBuildLogFollow(t *testing.T) {
76 mustGit(t, dir, env, "add", ".") 71 mustGit(t, dir, env, "add", ".")
77 mustGit(t, dir, env, "commit", "-q", "-m", "ci") 72 mustGit(t, dir, env, "commit", "-q", "-m", "ci")
78 mustGit(t, dir, env, "push", "-q", "origin", "main") 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.
78func claimBuild(t *testing.T, inst *instance, runnerKey string) string {
79 t.Helper()
81 out, errOut, code := inst.ssh(t, runnerKey, "", "runner", "next", "--json") 80 out, errOut, code := inst.ssh(t, runnerKey, "", "runner", "next", "--json")
82 if code != 0 { 81 if code != 0 {
83 t.Fatalf("runner next: %s", errOut) 82 t.Fatalf("runner next: %s", errOut)
@@ -90,7 +89,22 @@ func TestBuildLogFollow(t *testing.T) {
90 if err := json.Unmarshal([]byte(out), &claim); err != nil || claim.Data.ID == 0 { 89 if err := json.Unmarshal([]byte(out), &claim); err != nil || claim.Data.ID == 0 {
91 t.Fatalf("runner next output %q: %v", out, err) 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.
97func 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 cmd := inst.sshCmd(aliceKey, "build", "log", "alice/app", "1", "--follow") 109 cmd := inst.sshCmd(aliceKey, "build", "log", "alice/app", "1", "--follow")
96 stdout, err := cmd.StdoutPipe() 110 stdout, err := cmd.StdoutPipe()
e2e/shutdown_test.go +74
@@ -1,6 +1,7 @@
1package e2e 1package e2e
2 2
3import ( 3import (
4 "errors"
4 "fmt" 5 "fmt"
5 "net/http" 6 "net/http"
6 "os/exec" 7 "os/exec"
@@ -83,3 +84,76 @@ func TestShutdownClosesIdleConnections(t *testing.T) {
83 t.Fatalf("shutdown took %s with only an idle connection open", took) 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.
91func 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 if err != nil { 200 if err != nil {
201 return c.fail(protocol.ExitUsage, "%v", err) 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 if code >= 0 { 204 if code >= 0 {
205 return code 205 return code
206 } 206 }
207 if f.Has("--follow") { 207 if f.Has("--follow") {
208 return followBuildLog(c, b) 208 return followBuildLog(c, repo, b)
209 } 209 }
210 log, err := c.Store.BuildLog(b.ID) 210 log, err := c.Store.BuildLog(b.ID)
211 if err != nil { 211 if err != nil {
internal/control/buildfollow.go +74 −3
@@ -1,10 +1,12 @@
1package control 1package control
2 2
3import ( 3import (
4 "errors"
4 "fmt" 5 "fmt"
5 "sync" 6 "sync"
6 "time" 7 "time"
7 8
9 "gitbay.org/gitbay/internal/policy"
8 "gitbay.org/gitbay/internal/protocol" 10 "gitbay.org/gitbay/internal/protocol"
9 "gitbay.org/gitbay/internal/store" 11 "gitbay.org/gitbay/internal/store"
10) 12)
@@ -28,6 +30,11 @@ var (
28 // reaper's deadline, so a follow needs its own limit for the queued 30 // reaper's deadline, so a follow needs its own limit for the queued
29 // case or it never ends. 31 // case or it never ends.
30 followQueued = 10 * time.Minute 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
33var ( 40var (
@@ -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.
67func 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.
98func 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// followBuildLog writes the build's log as it grows and returns once the 107// followBuildLog writes the build's log as it grows and returns once the
57// build has an outcome and its last bytes are written. The outcome goes 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. 109// to stderr, so stdout is the log byte for byte. Read access is checked
59func followBuildLog(c *Ctx, b store.Build) int { 110// again every followPoll: a repository made private, or a grant revoked,
111// ends the follow with the answer a new request would get.
112func followBuildLog(c *Ctx, repo store.Repo, b store.Build) int {
60 if !takeFollow(c.User.ID) { 113 if !takeFollow(c.User.ID) {
61 return c.fail(protocol.ExitDenied, "%d follows are already open for this account; close one and retry", maxFollows) 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 var off int64 118 var off int64
66 var settleBy time.Time 119 var settleBy time.Time
67 var queuedSince time.Time 120 var queuedSince time.Time
121 checked := time.Now()
68 for { 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 wake := c.Store.BuildLogWait(b.ID) 133 wake := c.Store.BuildLogWait(b.ID)
70 status, chunk, err := c.Store.BuildLogFrom(b.ID, off) 134 status, chunk, err := c.Store.BuildLogFrom(b.ID, off)
71 if err != nil { 135 if err != nil {
@@ -107,10 +171,17 @@ func followBuildLog(c *Ctx, b store.Build) int {
107 t := time.NewTimer(wait) 171 t := time.NewTimer(wait)
108 select { 172 select {
109 case <-wake: 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 case <-t.C: 181 case <-t.C:
111 case <-c.Done: 182 case <-c.Done:
112 t.Stop() 183 t.Stop()
113 return protocol.ExitFailure 184 return ended(c)
114 } 185 }
115 t.Stop() 186 t.Stop()
116 } 187 }
internal/control/buildfollow_test.go +103 −13
@@ -33,6 +33,13 @@ func (b *syncBuffer) String() string {
33// follow starts build log --follow on build 1 of repo and returns the 33// follow starts build log --follow on build 1 of repo and returns the
34// buffers and a channel carrying the exit code. 34// buffers and a channel carrying the exit code.
35func follow(t *testing.T, st *store.Store, uid int64, repo store.Repo, done <-chan struct{}) (*syncBuffer, *syncBuffer, chan int) { 35func 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.
42func followStopping(t *testing.T, st *store.Store, uid int64, repo store.Repo, done, stopping <-chan struct{}) (*syncBuffer, *syncBuffer, chan int) {
36 t.Helper() 43 t.Helper()
37 u, err := st.UserByID(uid) 44 u, err := st.UserByID(uid)
38 if err != nil { 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 var out, errOut syncBuffer 48 var out, errOut syncBuffer
42 c := &Ctx{User: u, Scope: "full", Store: st, Stdin: strings.NewReader(""), 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 res := make(chan int, 1) 51 res := make(chan int, 1)
45 go func() { res <- Dispatch(c, []string{"build", "log", repo.Path(), "1", "--follow"}) }() 52 go func() { res <- Dispatch(c, []string{"build", "log", repo.Path(), "1", "--follow"}) }()
46 return &out, &errOut, res 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.
58func 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
49func waitExit(t *testing.T, res chan int) int { 70func waitExit(t *testing.T, res chan int) int {
50 t.Helper() 71 t.Helper()
51 select { 72 select {
@@ -58,9 +79,9 @@ func waitExit(t *testing.T, res chan int) int {
58} 79}
59 80
60func shortFollowTimers(t *testing.T) { 81func shortFollowTimers(t *testing.T) {
61 settle, poll, queued := followSettle, followPoll, followQueued 82 settle, poll, queued, coalesce := followSettle, followPoll, followQueued, followCoalesce
62 followSettle, followPoll = 200*time.Millisecond, 50*time.Millisecond 83 followSettle, followPoll, followCoalesce = 200*time.Millisecond, 50*time.Millisecond, 10*time.Millisecond
63 t.Cleanup(func() { followSettle, followPoll, followQueued = settle, poll, queued }) 84 t.Cleanup(func() { followSettle, followPoll, followQueued, followCoalesce = settle, poll, queued, coalesce })
64} 85}
65 86
66// The follow prints the stored log, then what arrives, and ends with the 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 st.AppendBuildLog(id, []byte("step one\n")) 154 st.AppendBuildLog(id, []byte("step one\n"))
134 done := make(chan struct{}) 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) 158 waitOutput(t, out, "step one")
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 }
145 close(done) 159 close(done)
146 if code := waitExit(t, res); code != protocol.ExitFailure { 160 if code := waitExit(t, res); code != protocol.ExitFailure {
147 t.Fatalf("exit %d, want %d", code, protocol.ExitFailure) 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.
170func 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.
192func 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// A build that stays pending ends its own follow: nothing reaps a queued 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.
234func 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// An account holding maxFollows is refused another. 259// An account holding maxFollows is refused another.
170func TestBuildLogFollowCap(t *testing.T) { 260func TestBuildLogFollowCap(t *testing.T) {
171 st, repo, uid := newQueueTestRepo(t) 261 st, repo, uid := newQueueTestRepo(t)
internal/control/control.go +4
@@ -44,6 +44,10 @@ type Ctx struct {
44 // more: the SSH channel closed or the HTTP request ended. A command 44 // more: the SSH channel closed or the HTTP request ended. A command
45 // that runs until something happens (build log --follow) stops on it. 45 // that runs until something happens (build log --follow) stops on it.
46 Done <-chan struct{} 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// usage reports a bad invocation with the command's registered usage, 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 JSON: true, 73 JSON: true,
74 ViaAPI: true, 74 ViaAPI: true,
75 ReadOnly: scope == "read", 75 ReadOnly: scope == "read",
76 Done: r.Context().Done(), 76 Done: s.until(r),
77 Stopping: s.stopping,
77 } 78 }
78 code := control.Dispatch(ctx, req.Argv) 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 JSON: true, 62 JSON: true,
63 ViaAPI: true, 63 ViaAPI: true,
64 ReadOnly: true, 64 ReadOnly: true,
65 Done: r.Context().Done(), 65 Done: s.until(r),
66 Stopping: s.stopping,
66 } 67 }
67 code := control.Dispatch(ctx, argv) 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 path := v.Repo.Path() 354 path := v.Repo.Path()
355 msg, code := s.runControlStream(viewer, []string{"build", "log", path, n, "--follow"}, 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 if r.Context().Err() != nil { 357 if r.Context().Err() != nil {
358 // The client left; nothing more to write. 358 // The client left; nothing more to write.
359 return 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
57func (s *Server) runControlStream(u store.User, argv []string, out io.Writer, done <-chan struct{}) (msg string, code int) { 57func (s *Server) runControlStream(u store.User, argv []string, out io.Writer, done <-chan struct{}) (msg string, code int) {
58 var stderr bytes.Buffer 58 var stderr bytes.Buffer
59 ctx := &control.Ctx{ 59 ctx := &control.Ctx{
60 User: u, 60 User: u,
61 Source: "web", 61 Source: "web",
62 Scope: "full", 62 Scope: "full",
63 Store: s.st, 63 Store: s.st,
64 Cfg: s.cfg, 64 Cfg: s.cfg,
65 Stdin: strings.NewReader(""), 65 Stdin: strings.NewReader(""),
66 Stdout: out, 66 Stdout: out,
67 Stderr: &stderr, 67 Stderr: &stderr,
68 ViaAPI: true, 68 ViaAPI: true,
69 Done: done, 69 Done: done,
70 Stopping: s.stopping,
70 } 71 }
71 code = control.Dispatch(ctx, argv) 72 code = control.Dispatch(ctx, argv)
72 return strings.TrimSpace(stderr.String()), code 73 return strings.TrimSpace(stderr.String()), code
internal/httpd/smart.go +27 −2
@@ -14,6 +14,7 @@ import (
14 "os" 14 "os"
15 "os/exec" 15 "os/exec"
16 "strings" 16 "strings"
17 "sync"
17 18
18 "gitbay.org/gitbay/internal/config" 19 "gitbay.org/gitbay/internal/config"
19 "gitbay.org/gitbay/internal/control" 20 "gitbay.org/gitbay/internal/control"
@@ -25,12 +26,36 @@ type Server struct {
25 cfg config.Config 26 cfg config.Config
26 st *store.Store 27 st *store.Store
27 apiLimit *apiLimiter 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
31func New(cfg config.Config, st *store.Store) *Server { 34func New(cfg config.Config, st *store.Store) *Server {
32 proxies, _ := cfg.HTTP.TrustedProxyNets() // validated at config load 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.
43func (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.
49func (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// receivePackRefusal exists only to fail legibly if a client POSTs without 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 sessions sync.WaitGroup // accepted connections still being served 39 sessions sync.WaitGroup // accepted connections still being served
40 mu sync.Mutex 40 mu sync.Mutex
41 conns map[*conn]struct{} 41 conns map[*conn]struct{}
42 stopping chan struct{} // closed by Stop
43 stopOnce sync.Once
42} 44}
43 45
44// conn is one accepted connection and how many sessions it is running. 46// conn is one accepted connection and how many sessions it is running.
@@ -51,7 +53,7 @@ type conn struct {
51} 53}
52 54
53func New(cfg config.Config, st *store.Store) (*Server, error) { 55func 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 sc := &ssh.ServerConfig{ 58 sc := &ssh.ServerConfig{
57 PublicKeyCallback: s.authenticate, 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.
185func (s *Server) Stop() {
186 s.stopOnce.Do(func() { close(s.stopping) })
187}
188
180// Shutdown closes every idle connection, then waits for the ones with a 189// Shutdown closes every idle connection, then waits for the ones with a
181// session running, or for ctx. The caller closes the listener first; a 190// session running, or for ctx. The caller closes the listener first; a
182// push in flight completes rather than being cut mid-pack. 191// push in flight completes rather than being cut mid-pack.
183func (s *Server) Shutdown(ctx context.Context) error { 192func (s *Server) Shutdown(ctx context.Context) error {
193 s.Stop()
184 s.mu.Lock() 194 s.mu.Lock()
185 for c := range s.conns { 195 for c := range s.conns {
186 if c.active.Load() == 0 { 196 if c.active.Load() == 0 {
@@ -240,12 +250,21 @@ func (s *Server) handleSession(sconn *ssh.ServerConn, ch ssh.Channel, reqs <-cha
240 req.Reply(true, nil) 250 req.Reply(true, nil)
241 // x/crypto closes reqs when the client closes the channel. That 251 // x/crypto closes reqs when the client closes the channel. That
242 // is how a follow learns nobody is reading: the CLI's shared 252 // is how a follow learns nobody is reading: the CLI's shared
243 // connection outlives a Ctrl-C, the channel does not. 253 // connection outlives a Ctrl-C, the channel does not. Stop
244 done := make(chan struct{}) 254 // ends it too, for a restart.
255 closed := make(chan struct{})
245 go func() { 256 go func() {
246 for r := range reqs { 257 for r := range reqs {
247 r.Reply(false, nil) 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 close(done) 268 close(done)
250 }() 269 }()
251 code := s.runExec(sconn, ch, payload.Command, done) 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 return protocol.ExitDenied 302 return protocol.ExitDenied
284 } 303 }
285 _ = s.st.TouchSSHKey(keyID) 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// runAnonymous handles a session from an unregistered key: the register 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// single dispatch path shared by the embedded listener and the system-sshd 333// single dispatch path shared by the embedded listener and the system-sshd
315// forced command (gitbayd shell). 334// forced command (gitbayd shell).
316func Exec(cfg config.Config, st *store.Store, user store.User, scope, source, cmdline string, 335func 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 if user.Disabled { 337 if user.Disabled {
319 fmt.Fprintln(stderr, "this account is disabled; contact the instance admin") 338 fmt.Fprintln(stderr, "this account is disabled; contact the instance admin")
320 return protocol.ExitDenied 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 ctx := &control.Ctx{ 364 ctx := &control.Ctx{
346 User: user, 365 User: user,
347 Scope: scope, 366 Scope: scope,
348 Source: source, 367 Source: source,
349 Store: st, 368 Store: st,
350 Cfg: cfg, 369 Cfg: cfg,
351 Stdin: stdin, 370 Stdin: stdin,
352 Stdout: stdout, 371 Stdout: stdout,
353 Stderr: stderr, 372 Stderr: stderr,
354 Done: done, 373 Done: done,
374 Stopping: stopping,
355 } 375 }
356 return control.Dispatch(ctx, argv) 376 return control.Dispatch(ctx, argv)
357} 377}
internal/sshd/sshd_test.go added +199
@@ -0,0 +1,199 @@
1package sshd
2
3import (
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.
24func 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.
91func 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
122func 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.
135func 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.
158func 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.
179func 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}