fix(ci): follow waits for lines on a running job and a busy stream revalidates
GetJobLog answered out_of_range to a follower that asked for a line the runner had not written yet; out_of_range and not_found are now the answer only without follow or for a job that ended, as the contract says. A stream that always had lines to send never asked again whether its token still existed; the check now runs before each send. The web live feed ends cleanly for a job that ended without a log.
- commit
- b7f6252061831da1d17c33264e321458d5aaf5fe
- author
- brice · · 2026-10-12 19:12 UTC
- parent
- 008a3aa
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
3 files · +149 -18
M internal/adapters/web/actions.go +1 -1
| @@ -818,7 +818,7 @@ func (s *Server) logStream(w http.ResponseWriter, r *http.Request) { | |||
| 818 | 818 | } | |
| 819 | 819 | return rc.Flush() | |
| 820 | 820 | }) | |
| 821 | - | if err == nil { | |
| 821 | + | if err == nil || errors.Is(err, domain.ErrNotFound) { | |
| 822 | 822 | _ = out.raw("event: ci-end\nid: " + strconv.FormatInt(n, 10) + "\ndata: end\n\n") | |
| 823 | 823 | } | |
| 824 | 824 | } | |
| @@ -818,7 +818,7 @@ func (s *Server) logStream(w http.ResponseWriter, r *http.Request) { | |||||
| 818 | } | 818 | } | ||
| 819 | return rc.Flush() | 819 | return rc.Flush() | ||
| 820 | }) | 820 | }) | ||
| 821 | - | if err == nil { | 821 | + | if err == nil || errors.Is(err, domain.ErrNotFound) { |
| 822 | _ = out.raw("event: ci-end\nid: " + strconv.FormatInt(n, 10) + "\ndata: end\n\n") | 822 | _ = out.raw("event: ci-end\nid: " + strconv.FormatInt(n, 10) + "\ndata: end\n\n") | ||
| 823 | } | 823 | } | ||
| 824 | } | 824 | } | ||
M internal/usecase/ci_runner_unit_test.go +121 -0
| @@ -148,3 +148,124 @@ func TestLogLimitOnAVanishedJobIsNotFound(t *testing.T) { | |||
| 148 | 148 | _, isEnded := domain.IsJobEnded(err) | |
| 149 | 149 | assert.False(t, isEnded) | |
| 150 | 150 | } | |
| 151 | + | ||
| 152 | + | // expectRunJob makes GetJobLog find the job of the repository (a public one). | |
| 153 | + | func (u *ciUnit) expectLogJob(j domain.CIJob) { | |
| 154 | + | u.meta.EXPECT().Get(mock.Anything, j.Repo).Return(domain.Repo{Name: j.Repo, Visibility: domain.VisibilityPublic}, nil) | |
| 155 | + | u.store.EXPECT().GetRun(mock.Anything, j.Repo, j.RunID).Return(domain.CIRun{Repo: j.Repo, ID: j.RunID, Attempt: 1}, nil) | |
| 156 | + | u.store.EXPECT().GetJob(mock.Anything, j.RunID, j.RunAttempt, j.ID).Return(j, nil) | |
| 157 | + | } | |
| 158 | + | ||
| 159 | + | // S7: without follow a missing log is not found and an offset past the end is | |
| 160 | + | // out of range; with follow a running job's reader waits for the lines, and a | |
| 161 | + | // job that ended answers as without follow (follow changes nothing then). | |
| 162 | + | func TestGetJobLogFollowAndOutOfRange(t *testing.T) { | |
| 163 | + | t.Parallel() | |
| 164 | + | p := domain.Anonymous() | |
| 165 | + | params := func(follow bool, offset int64) ports.GetJobLogParams { | |
| 166 | + | return ports.GetJobLogParams{JobID: jobIDOne, Offset: offset, Follow: follow} | |
| 167 | + | } | |
| 168 | + | send := func(ports.LogBatch) error { return nil } | |
| 169 | + | ||
| 170 | + | t.Run("no follow: out of range", func(t *testing.T) { | |
| 171 | + | t.Parallel() | |
| 172 | + | u := newCIUnit(t, nil) | |
| 173 | + | j := ciJob(t, domain.CIPhaseRunning, jobSecret) | |
| 174 | + | u.expectLogJob(j) | |
| 175 | + | u.logs.EXPECT().Changed(mock.Anything).Return(make(chan struct{})) | |
| 176 | + | u.logs.EXPECT().Read(mock.Anything, mock.Anything, int64(5), mock.Anything, mock.Anything).Return(nil, domain.ErrOutOfRange) | |
| 177 | + | err := u.ci.GetJobLog(context.Background(), p, j.Repo, params(false, 5), send) | |
| 178 | + | assert.ErrorIs(t, err, domain.ErrOutOfRange) | |
| 179 | + | }) | |
| 180 | + | t.Run("no follow: no log", func(t *testing.T) { | |
| 181 | + | t.Parallel() | |
| 182 | + | u := newCIUnit(t, nil) | |
| 183 | + | j := ciJob(t, domain.CIPhaseReady, jobSecret) | |
| 184 | + | u.expectLogJob(j) | |
| 185 | + | u.logs.EXPECT().Changed(mock.Anything).Return(make(chan struct{})) | |
| 186 | + | u.logs.EXPECT().Read(mock.Anything, mock.Anything, int64(0), mock.Anything, mock.Anything).Return(nil, domain.ErrNotFound) | |
| 187 | + | assert.ErrorIs(t, u.ci.GetJobLog(context.Background(), p, j.Repo, params(false, 0), send), domain.ErrNotFound) | |
| 188 | + | }) | |
| 189 | + | t.Run("follow, ended: out of range like without follow", func(t *testing.T) { | |
| 190 | + | t.Parallel() | |
| 191 | + | u := newCIUnit(t, nil) | |
| 192 | + | j := ciJob(t, domain.CIPhaseEnded, jobSecret) | |
| 193 | + | u.expectLogJob(j) | |
| 194 | + | u.logs.EXPECT().Changed(mock.Anything).Return(make(chan struct{})) | |
| 195 | + | u.logs.EXPECT().Read(mock.Anything, mock.Anything, int64(5), mock.Anything, mock.Anything).Return(nil, domain.ErrOutOfRange) | |
| 196 | + | err := u.ci.GetJobLog(context.Background(), p, j.Repo, params(true, 5), send) | |
| 197 | + | assert.ErrorIs(t, err, domain.ErrOutOfRange) | |
| 198 | + | }) | |
| 199 | + | t.Run("follow, running: waits for the line, then sends it", func(t *testing.T) { | |
| 200 | + | t.Parallel() | |
| 201 | + | u := newCIUnit(t, nil) | |
| 202 | + | j := ciJob(t, domain.CIPhaseRunning, jobSecret) | |
| 203 | + | u.expectLogJob(j) | |
| 204 | + | wake := make(chan struct{}) | |
| 205 | + | u.logs.EXPECT().Changed(mock.Anything).Return((<-chan struct{})(wake)) | |
| 206 | + | reads := 0 | |
| 207 | + | u.logs.EXPECT().Read(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn( | |
| 208 | + | func(_ context.Context, _ ports.CILogRef, from int64, _, _ int) ([]domain.CILogLine, error) { | |
| 209 | + | reads++ | |
| 210 | + | if from == 0 && reads == 1 { | |
| 211 | + | return nil, domain.ErrOutOfRange // the runner has not written yet | |
| 212 | + | } | |
| 213 | + | if from == 0 { | |
| 214 | + | return []domain.CILogLine{{Time: now, Text: "late line"}}, nil | |
| 215 | + | } | |
| 216 | + | return nil, nil | |
| 217 | + | }) | |
| 218 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 219 | + | got := make(chan ports.LogBatch, 1) | |
| 220 | + | errc := make(chan error, 1) | |
| 221 | + | go func() { | |
| 222 | + | errc <- u.ci.GetJobLog(ctx, p, j.Repo, params(true, 0), func(b ports.LogBatch) error { | |
| 223 | + | got <- b | |
| 224 | + | cancel() // enough | |
| 225 | + | return nil | |
| 226 | + | }) | |
| 227 | + | }() | |
| 228 | + | select { | |
| 229 | + | case <-got: | |
| 230 | + | t.Fatal("a line was sent before it existed") | |
| 231 | + | case <-time.After(150 * time.Millisecond): | |
| 232 | + | } | |
| 233 | + | close(wake) // the append happens | |
| 234 | + | select { | |
| 235 | + | case b := <-got: | |
| 236 | + | require.Len(t, b.Lines, 1) | |
| 237 | + | assert.Equal(t, "late line", b.Lines[0].Text) | |
| 238 | + | case <-time.After(5 * time.Second): | |
| 239 | + | t.Fatal("the reader did not wake up") | |
| 240 | + | } | |
| 241 | + | require.NoError(t, <-errc) | |
| 242 | + | }) | |
| 243 | + | } | |
| 244 | + | ||
| 245 | + | // S8: a stream that never idles revalidates the token too. | |
| 246 | + | func TestGetJobLogRevalidatesABusyStream(t *testing.T) { | |
| 247 | + | t.Parallel() | |
| 248 | + | auth := mocks.NewRevalidator(t) | |
| 249 | + | u := newCIUnit(t, func(cfg *usecase.CIConfig) { cfg.Auth = auth }) | |
| 250 | + | j := ciJob(t, domain.CIPhaseRunning, jobSecret) | |
| 251 | + | u.expectLogJob(j) | |
| 252 | + | u.logs.EXPECT().Changed(mock.Anything).Return(make(chan struct{})) | |
| 253 | + | // A token principal; the log keeps having lines to send. | |
| 254 | + | tok := domain.TokenPrincipal(domain.Token{ID: "t1", Scopes: []domain.Scope{domain.ScopeRepoRead}}) | |
| 255 | + | u.logs.EXPECT().Read(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]domain.CILogLine{{Time: now, Text: "x"}}, nil) | |
| 256 | + | revoked := domain.ErrUnauthenticated | |
| 257 | + | auth.EXPECT().Revalidate(mock.Anything, tok).Return(revoked).Once() | |
| 258 | + | ||
| 259 | + | sent := 0 | |
| 260 | + | start := time.Now() | |
| 261 | + | err := u.ci.GetJobLog(context.Background(), tok, j.Repo, ports.GetJobLogParams{JobID: jobIDOne, Follow: true}, func(ports.LogBatch) error { | |
| 262 | + | sent++ | |
| 263 | + | time.Sleep(usecase.DefaultRevalidateEvery / 2) | |
| 264 | + | if time.Since(start) > 30*time.Second { | |
| 265 | + | return context.DeadlineExceeded | |
| 266 | + | } | |
| 267 | + | return nil | |
| 268 | + | }) | |
| 269 | + | assert.ErrorIs(t, err, domain.ErrUnauthenticated, "a revoked token stops a busy stream") | |
| 270 | + | assert.GreaterOrEqual(t, sent, 2) | |
| 271 | + | } | |
| @@ -148,3 +148,124 @@ func TestLogLimitOnAVanishedJobIsNotFound(t *testing.T) { | |||||
| 148 | _, isEnded := domain.IsJobEnded(err) | 148 | _, isEnded := domain.IsJobEnded(err) | ||
| 149 | assert.False(t, isEnded) | 149 | assert.False(t, isEnded) | ||
| 150 | } | 150 | } | ||
| 151 | + | ||||
| 152 | + | // expectRunJob makes GetJobLog find the job of the repository (a public one). | |||
| 153 | + | func (u *ciUnit) expectLogJob(j domain.CIJob) { | |||
| 154 | + | u.meta.EXPECT().Get(mock.Anything, j.Repo).Return(domain.Repo{Name: j.Repo, Visibility: domain.VisibilityPublic}, nil) | |||
| 155 | + | u.store.EXPECT().GetRun(mock.Anything, j.Repo, j.RunID).Return(domain.CIRun{Repo: j.Repo, ID: j.RunID, Attempt: 1}, nil) | |||
| 156 | + | u.store.EXPECT().GetJob(mock.Anything, j.RunID, j.RunAttempt, j.ID).Return(j, nil) | |||
| 157 | + | } | |||
| 158 | + | ||||
| 159 | + | // S7: without follow a missing log is not found and an offset past the end is | |||
| 160 | + | // out of range; with follow a running job's reader waits for the lines, and a | |||
| 161 | + | // job that ended answers as without follow (follow changes nothing then). | |||
| 162 | + | func TestGetJobLogFollowAndOutOfRange(t *testing.T) { | |||
| 163 | + | t.Parallel() | |||
| 164 | + | p := domain.Anonymous() | |||
| 165 | + | params := func(follow bool, offset int64) ports.GetJobLogParams { | |||
| 166 | + | return ports.GetJobLogParams{JobID: jobIDOne, Offset: offset, Follow: follow} | |||
| 167 | + | } | |||
| 168 | + | send := func(ports.LogBatch) error { return nil } | |||
| 169 | + | ||||
| 170 | + | t.Run("no follow: out of range", func(t *testing.T) { | |||
| 171 | + | t.Parallel() | |||
| 172 | + | u := newCIUnit(t, nil) | |||
| 173 | + | j := ciJob(t, domain.CIPhaseRunning, jobSecret) | |||
| 174 | + | u.expectLogJob(j) | |||
| 175 | + | u.logs.EXPECT().Changed(mock.Anything).Return(make(chan struct{})) | |||
| 176 | + | u.logs.EXPECT().Read(mock.Anything, mock.Anything, int64(5), mock.Anything, mock.Anything).Return(nil, domain.ErrOutOfRange) | |||
| 177 | + | err := u.ci.GetJobLog(context.Background(), p, j.Repo, params(false, 5), send) | |||
| 178 | + | assert.ErrorIs(t, err, domain.ErrOutOfRange) | |||
| 179 | + | }) | |||
| 180 | + | t.Run("no follow: no log", func(t *testing.T) { | |||
| 181 | + | t.Parallel() | |||
| 182 | + | u := newCIUnit(t, nil) | |||
| 183 | + | j := ciJob(t, domain.CIPhaseReady, jobSecret) | |||
| 184 | + | u.expectLogJob(j) | |||
| 185 | + | u.logs.EXPECT().Changed(mock.Anything).Return(make(chan struct{})) | |||
| 186 | + | u.logs.EXPECT().Read(mock.Anything, mock.Anything, int64(0), mock.Anything, mock.Anything).Return(nil, domain.ErrNotFound) | |||
| 187 | + | assert.ErrorIs(t, u.ci.GetJobLog(context.Background(), p, j.Repo, params(false, 0), send), domain.ErrNotFound) | |||
| 188 | + | }) | |||
| 189 | + | t.Run("follow, ended: out of range like without follow", func(t *testing.T) { | |||
| 190 | + | t.Parallel() | |||
| 191 | + | u := newCIUnit(t, nil) | |||
| 192 | + | j := ciJob(t, domain.CIPhaseEnded, jobSecret) | |||
| 193 | + | u.expectLogJob(j) | |||
| 194 | + | u.logs.EXPECT().Changed(mock.Anything).Return(make(chan struct{})) | |||
| 195 | + | u.logs.EXPECT().Read(mock.Anything, mock.Anything, int64(5), mock.Anything, mock.Anything).Return(nil, domain.ErrOutOfRange) | |||
| 196 | + | err := u.ci.GetJobLog(context.Background(), p, j.Repo, params(true, 5), send) | |||
| 197 | + | assert.ErrorIs(t, err, domain.ErrOutOfRange) | |||
| 198 | + | }) | |||
| 199 | + | t.Run("follow, running: waits for the line, then sends it", func(t *testing.T) { | |||
| 200 | + | t.Parallel() | |||
| 201 | + | u := newCIUnit(t, nil) | |||
| 202 | + | j := ciJob(t, domain.CIPhaseRunning, jobSecret) | |||
| 203 | + | u.expectLogJob(j) | |||
| 204 | + | wake := make(chan struct{}) | |||
| 205 | + | u.logs.EXPECT().Changed(mock.Anything).Return((<-chan struct{})(wake)) | |||
| 206 | + | reads := 0 | |||
| 207 | + | u.logs.EXPECT().Read(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn( | |||
| 208 | + | func(_ context.Context, _ ports.CILogRef, from int64, _, _ int) ([]domain.CILogLine, error) { | |||
| 209 | + | reads++ | |||
| 210 | + | if from == 0 && reads == 1 { | |||
| 211 | + | return nil, domain.ErrOutOfRange // the runner has not written yet | |||
| 212 | + | } | |||
| 213 | + | if from == 0 { | |||
| 214 | + | return []domain.CILogLine{{Time: now, Text: "late line"}}, nil | |||
| 215 | + | } | |||
| 216 | + | return nil, nil | |||
| 217 | + | }) | |||
| 218 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 219 | + | got := make(chan ports.LogBatch, 1) | |||
| 220 | + | errc := make(chan error, 1) | |||
| 221 | + | go func() { | |||
| 222 | + | errc <- u.ci.GetJobLog(ctx, p, j.Repo, params(true, 0), func(b ports.LogBatch) error { | |||
| 223 | + | got <- b | |||
| 224 | + | cancel() // enough | |||
| 225 | + | return nil | |||
| 226 | + | }) | |||
| 227 | + | }() | |||
| 228 | + | select { | |||
| 229 | + | case <-got: | |||
| 230 | + | t.Fatal("a line was sent before it existed") | |||
| 231 | + | case <-time.After(150 * time.Millisecond): | |||
| 232 | + | } | |||
| 233 | + | close(wake) // the append happens | |||
| 234 | + | select { | |||
| 235 | + | case b := <-got: | |||
| 236 | + | require.Len(t, b.Lines, 1) | |||
| 237 | + | assert.Equal(t, "late line", b.Lines[0].Text) | |||
| 238 | + | case <-time.After(5 * time.Second): | |||
| 239 | + | t.Fatal("the reader did not wake up") | |||
| 240 | + | } | |||
| 241 | + | require.NoError(t, <-errc) | |||
| 242 | + | }) | |||
| 243 | + | } | |||
| 244 | + | ||||
| 245 | + | // S8: a stream that never idles revalidates the token too. | |||
| 246 | + | func TestGetJobLogRevalidatesABusyStream(t *testing.T) { | |||
| 247 | + | t.Parallel() | |||
| 248 | + | auth := mocks.NewRevalidator(t) | |||
| 249 | + | u := newCIUnit(t, func(cfg *usecase.CIConfig) { cfg.Auth = auth }) | |||
| 250 | + | j := ciJob(t, domain.CIPhaseRunning, jobSecret) | |||
| 251 | + | u.expectLogJob(j) | |||
| 252 | + | u.logs.EXPECT().Changed(mock.Anything).Return(make(chan struct{})) | |||
| 253 | + | // A token principal; the log keeps having lines to send. | |||
| 254 | + | tok := domain.TokenPrincipal(domain.Token{ID: "t1", Scopes: []domain.Scope{domain.ScopeRepoRead}}) | |||
| 255 | + | u.logs.EXPECT().Read(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]domain.CILogLine{{Time: now, Text: "x"}}, nil) | |||
| 256 | + | revoked := domain.ErrUnauthenticated | |||
| 257 | + | auth.EXPECT().Revalidate(mock.Anything, tok).Return(revoked).Once() | |||
| 258 | + | ||||
| 259 | + | sent := 0 | |||
| 260 | + | start := time.Now() | |||
| 261 | + | err := u.ci.GetJobLog(context.Background(), tok, j.Repo, ports.GetJobLogParams{JobID: jobIDOne, Follow: true}, func(ports.LogBatch) error { | |||
| 262 | + | sent++ | |||
| 263 | + | time.Sleep(usecase.DefaultRevalidateEvery / 2) | |||
| 264 | + | if time.Since(start) > 30*time.Second { | |||
| 265 | + | return context.DeadlineExceeded | |||
| 266 | + | } | |||
| 267 | + | return nil | |||
| 268 | + | }) | |||
| 269 | + | assert.ErrorIs(t, err, domain.ErrUnauthenticated, "a revoked token stops a busy stream") | |||
| 270 | + | assert.GreaterOrEqual(t, sent, 2) | |||
| 271 | + | } | |||
M internal/usecase/ci_runs.go +27 -17
| @@ -218,7 +218,7 @@ func (c *CI) GetJobLog(ctx context.Context, p domain.Principal, name domain.Repo | |||
| 218 | 218 | if params.Offset < 0 { | |
| 219 | 219 | return fmt.Errorf("%w: negative offset", domain.ErrInvalid) | |
| 220 | 220 | } | |
| 221 | - | run, j, err := c.jobOf(ctx, name, params.JobID, params.Attempt) | |
| 221 | + | _, j, err := c.jobOf(ctx, name, params.JobID, params.Attempt) | |
| 222 | 222 | if err != nil { | |
| 223 | 223 | return err | |
| 224 | 224 | } | |
| @@ -234,6 +234,19 @@ func (c *CI) GetJobLog(ctx context.Context, p domain.Principal, name domain.Repo | |||
| 234 | 234 | jref := jobRef(j) | |
| 235 | 235 | offset := params.Offset | |
| 236 | 236 | checked := time.Now() | |
| 237 | + | // revalidate asks again whether the token of a stored-token caller still | |
| 238 | + | // exists, at most every few seconds, whether the stream is busy or idle: a | |
| 239 | + | // token revoked while a long log is sent stops the stream. | |
| 240 | + | revalidate := func() error { | |
| 241 | + | if p.TokenID() == "" || c.cfg.Auth == nil || time.Since(checked) <= DefaultRevalidateEvery { | |
| 242 | + | return nil | |
| 243 | + | } | |
| 244 | + | if err := c.cfg.Auth.Revalidate(ctx, p); err != nil { | |
| 245 | + | return err | |
| 246 | + | } | |
| 247 | + | checked = time.Now() | |
| 248 | + | return nil | |
| 249 | + | } | |
| 237 | 250 | for { | |
| 238 | 251 | wakeJob, wakeLog := c.jobs.wait(jref), c.cfg.Logs.Changed(ref) | |
| 239 | 252 | cur, err := c.cfg.Store.GetJob(ctx, j.RunID, j.RunAttempt, j.ID) | |
| @@ -243,24 +256,25 @@ func (c *CI) GetJobLog(ctx context.Context, p domain.Principal, name domain.Repo | |||
| 243 | 256 | done := cur.Ended() | |
| 244 | 257 | lines, err := c.cfg.Logs.Read(ctx, ref, offset, domain.CIMaxLogLines, domain.CIMaxLogChunkBytes) | |
| 245 | 258 | switch { | |
| 246 | - | case errors.Is(err, domain.ErrNotFound): | |
| 247 | - | // No line yet. Without follow, or for a job that ended without | |
| 248 | - | // lines, there is no log. | |
| 259 | + | case errors.Is(err, domain.ErrNotFound), errors.Is(err, domain.ErrOutOfRange): | |
| 260 | + | // No line yet (a queued job has no log), or fewer lines than the | |
| 261 | + | // offset. Without follow, and for a job that ended (follow changes | |
| 262 | + | // nothing then), that is the answer (contract: not_found, | |
| 263 | + | // out_of_range). A job that runs may still write the line: wait. | |
| 249 | 264 | if !params.Follow || done { | |
| 250 | - | if done && params.Offset == 0 && params.Follow { | |
| 251 | - | return nil | |
| 265 | + | if errors.Is(err, domain.ErrNotFound) { | |
| 266 | + | return domain.ErrNotFound | |
| 252 | 267 | } | |
| 253 | - | return domain.ErrNotFound | |
| 254 | - | } | |
| 255 | - | case errors.Is(err, domain.ErrOutOfRange): | |
| 256 | - | if !params.Follow || done { | |
| 257 | 268 | return fmt.Errorf("%w: the log has fewer than %d lines", domain.ErrOutOfRange, offset) | |
| 258 | 269 | } | |
| 259 | - | return fmt.Errorf("%w: the log has fewer than %d lines", domain.ErrOutOfRange, offset) | |
| 270 | + | lines = nil | |
| 260 | 271 | case err != nil: | |
| 261 | 272 | return err | |
| 262 | 273 | } | |
| 263 | 274 | if len(lines) > 0 { | |
| 275 | + | if err := revalidate(); err != nil { | |
| 276 | + | return err | |
| 277 | + | } | |
| 264 | 278 | batch := ports.LogBatch{Offset: offset, Lines: c.withSteps(ctx, cur, offset, lines)} | |
| 265 | 279 | if err := send(batch); err != nil { | |
| 266 | 280 | return err | |
| @@ -271,13 +285,9 @@ func (c *CI) GetJobLog(ctx context.Context, p domain.Principal, name domain.Repo | |||
| 271 | 285 | if !params.Follow || done { | |
| 272 | 286 | return nil | |
| 273 | 287 | } | |
| 274 | - | if p.TokenID() != "" && c.cfg.Auth != nil && time.Since(checked) > DefaultRevalidateEvery { | |
| 275 | - | if err := c.cfg.Auth.Revalidate(ctx, p); err != nil { | |
| 276 | - | return err | |
| 277 | - | } | |
| 278 | - | checked = time.Now() | |
| 288 | + | if err := revalidate(); err != nil { | |
| 289 | + | return err | |
| 279 | 290 | } | |
| 280 | - | _ = run | |
| 281 | 291 | select { | |
| 282 | 292 | case <-wakeJob: | |
| 283 | 293 | case <-wakeLog: | |
| @@ -218,7 +218,7 @@ func (c *CI) GetJobLog(ctx context.Context, p domain.Principal, name domain.Repo | |||||
| 218 | if params.Offset < 0 { | 218 | if params.Offset < 0 { | ||
| 219 | return fmt.Errorf("%w: negative offset", domain.ErrInvalid) | 219 | return fmt.Errorf("%w: negative offset", domain.ErrInvalid) | ||
| 220 | } | 220 | } | ||
| 221 | - | run, j, err := c.jobOf(ctx, name, params.JobID, params.Attempt) | 221 | + | _, j, err := c.jobOf(ctx, name, params.JobID, params.Attempt) |
| 222 | if err != nil { | 222 | if err != nil { | ||
| 223 | return err | 223 | return err | ||
| 224 | } | 224 | } | ||
| @@ -234,6 +234,19 @@ func (c *CI) GetJobLog(ctx context.Context, p domain.Principal, name domain.Repo | |||||
| 234 | jref := jobRef(j) | 234 | jref := jobRef(j) | ||
| 235 | offset := params.Offset | 235 | offset := params.Offset | ||
| 236 | checked := time.Now() | 236 | checked := time.Now() | ||
| 237 | + | // revalidate asks again whether the token of a stored-token caller still | |||
| 238 | + | // exists, at most every few seconds, whether the stream is busy or idle: a | |||
| 239 | + | // token revoked while a long log is sent stops the stream. | |||
| 240 | + | revalidate := func() error { | |||
| 241 | + | if p.TokenID() == "" || c.cfg.Auth == nil || time.Since(checked) <= DefaultRevalidateEvery { | |||
| 242 | + | return nil | |||
| 243 | + | } | |||
| 244 | + | if err := c.cfg.Auth.Revalidate(ctx, p); err != nil { | |||
| 245 | + | return err | |||
| 246 | + | } | |||
| 247 | + | checked = time.Now() | |||
| 248 | + | return nil | |||
| 249 | + | } | |||
| 237 | for { | 250 | for { | ||
| 238 | wakeJob, wakeLog := c.jobs.wait(jref), c.cfg.Logs.Changed(ref) | 251 | wakeJob, wakeLog := c.jobs.wait(jref), c.cfg.Logs.Changed(ref) | ||
| 239 | cur, err := c.cfg.Store.GetJob(ctx, j.RunID, j.RunAttempt, j.ID) | 252 | cur, err := c.cfg.Store.GetJob(ctx, j.RunID, j.RunAttempt, j.ID) | ||
| @@ -243,24 +256,25 @@ func (c *CI) GetJobLog(ctx context.Context, p domain.Principal, name domain.Repo | |||||
| 243 | done := cur.Ended() | 256 | done := cur.Ended() | ||
| 244 | lines, err := c.cfg.Logs.Read(ctx, ref, offset, domain.CIMaxLogLines, domain.CIMaxLogChunkBytes) | 257 | lines, err := c.cfg.Logs.Read(ctx, ref, offset, domain.CIMaxLogLines, domain.CIMaxLogChunkBytes) | ||
| 245 | switch { | 258 | switch { | ||
| 246 | - | case errors.Is(err, domain.ErrNotFound): | 259 | + | case errors.Is(err, domain.ErrNotFound), errors.Is(err, domain.ErrOutOfRange): |
| 247 | - | // No line yet. Without follow, or for a job that ended without | 260 | + | // No line yet (a queued job has no log), or fewer lines than the |
| 248 | - | // lines, there is no log. | 261 | + | // offset. Without follow, and for a job that ended (follow changes |
| 262 | + | // nothing then), that is the answer (contract: not_found, | |||
| 263 | + | // out_of_range). A job that runs may still write the line: wait. | |||
| 249 | if !params.Follow || done { | 264 | if !params.Follow || done { | ||
| 250 | - | if done && params.Offset == 0 && params.Follow { | 265 | + | if errors.Is(err, domain.ErrNotFound) { |
| 251 | - | return nil | 266 | + | return domain.ErrNotFound |
| 252 | } | 267 | } | ||
| 253 | - | return domain.ErrNotFound | |||
| 254 | - | } | |||
| 255 | - | case errors.Is(err, domain.ErrOutOfRange): | |||
| 256 | - | if !params.Follow || done { | |||
| 257 | return fmt.Errorf("%w: the log has fewer than %d lines", domain.ErrOutOfRange, offset) | 268 | return fmt.Errorf("%w: the log has fewer than %d lines", domain.ErrOutOfRange, offset) | ||
| 258 | } | 269 | } | ||
| 259 | - | return fmt.Errorf("%w: the log has fewer than %d lines", domain.ErrOutOfRange, offset) | 270 | + | lines = nil |
| 260 | case err != nil: | 271 | case err != nil: | ||
| 261 | return err | 272 | return err | ||
| 262 | } | 273 | } | ||
| 263 | if len(lines) > 0 { | 274 | if len(lines) > 0 { | ||
| 275 | + | if err := revalidate(); err != nil { | |||
| 276 | + | return err | |||
| 277 | + | } | |||
| 264 | batch := ports.LogBatch{Offset: offset, Lines: c.withSteps(ctx, cur, offset, lines)} | 278 | batch := ports.LogBatch{Offset: offset, Lines: c.withSteps(ctx, cur, offset, lines)} | ||
| 265 | if err := send(batch); err != nil { | 279 | if err := send(batch); err != nil { | ||
| 266 | return err | 280 | return err | ||
| @@ -271,13 +285,9 @@ func (c *CI) GetJobLog(ctx context.Context, p domain.Principal, name domain.Repo | |||||
| 271 | if !params.Follow || done { | 285 | if !params.Follow || done { | ||
| 272 | return nil | 286 | return nil | ||
| 273 | } | 287 | } | ||
| 274 | - | if p.TokenID() != "" && c.cfg.Auth != nil && time.Since(checked) > DefaultRevalidateEvery { | 288 | + | if err := revalidate(); err != nil { |
| 275 | - | if err := c.cfg.Auth.Revalidate(ctx, p); err != nil { | 289 | + | return err |
| 276 | - | return err | |||
| 277 | - | } | |||
| 278 | - | checked = time.Now() | |||
| 279 | } | 290 | } | ||
| 280 | - | _ = run | |||
| 281 | select { | 291 | select { | ||
| 282 | case <-wakeJob: | 292 | case <-wakeJob: | ||
| 283 | case <-wakeLog: | 293 | case <-wakeLog: | ||