fix(web): count the log pages and the raw log in the per-client stream limit

Only the live feed named the visitor; the job page and the raw log passed no
client key to GetJobLog, so they escaped the limit of streams per client.

commit: 5e9332cb60a4b9d5b65cfddb816d53ef8ef9c364
author: brice · 9d ago · 2026-10-12 19:13 UTC
parent: 8dbcdb5
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +80 -5

M internal/adapters/web/actions.go +10 -5


@@ -422,7 +422,7 @@ func (s Server) job(ctx context.Context, r http.Request) (templ.Component, int
422	422		}
423	423		p.Follow = !view.Ended()
424	424		p.StreamHref = jobHref(d.name(), id, jobID) + "/stream"
425		-	lines, next, err := s.readLog(ctx, d.repo.Name, view, from, jobLogPageLines)
	425	+	lines, next, err := s.readLog(ctx, s.logClient(r), d.repo.Name, view, from, jobLogPageLines)
426	426		if err != nil {
427	427		return nil, 0, err
428	428		}

@@ -452,9 +452,14 @@ func (s Server) job(ctx context.Context, r http.Request) (templ.Component, int
452	452		return Job(l, p), http.StatusOK, nil
453	453		}
454	454
	455	+	// logClient is the key of the visitor for the limit of log streams per client.
	456	+	func (s Server) logClient(r http.Request) string {
	457	+	return clientlimit.Key(clientlimit.ClientAddr(r, s.cfg.TrustedProxies))
	458	+	}
	459	+
455	460		// readLog reads up to max lines from a line number, through the log stream of
456	461		// the use case, and returns the line after the last read.
457		-	func (s *Server) readLog(ctx context.Context, repo domain.RepoName, j ports.JobView, from int64, max int) ([]LogLineView, int64, error) {
	462	+	func (s *Server) readLog(ctx context.Context, client string, repo domain.RepoName, j ports.JobView, from int64, max int) ([]LogLineView, int64, error) {
458	463		if from > j.Lines \|\| j.Lines == 0 {
459	464		return nil, from, nil
460	465		}

@@ -470,7 +475,7 @@ func (s *Server) readLog(ctx context.Context, repo domain.RepoName, j ports.JobV
470	475		return 0
471	476		}
472	477		stop := errors.New("enough")
473		-	err := s.cfg.Runs.GetJobLog(ctx, principal(ctx), repo, ports.GetJobLogParams{JobID: j.ID, Attempt: j.Attempt, Offset: from}, func(b ports.LogBatch) error {
	478	+	err := s.cfg.Runs.GetJobLog(ctx, principal(ctx), repo, ports.GetJobLogParams{JobID: j.ID, Attempt: j.Attempt, Offset: from, Client: client}, func(b ports.LogBatch) error {
474	479		for i, l := range b.Lines {
475	480		if len(out) >= max {
476	481		return stop

@@ -541,7 +546,7 @@ func (s Server) rawLog(w http.ResponseWriter, r http.Request) {
541	546		cacheHeaders(ctx, h, "no-cache", sessionCache)
542	547		written := 0
543	548		open := false
544		-	err = s.cfg.Runs.GetJobLog(ctx, principal(ctx), d.repo.Name, ports.GetJobLogParams{JobID: jobID, Attempt: attempt}, func(b ports.LogBatch) error {
	549	+	err = s.cfg.Runs.GetJobLog(ctx, principal(ctx), d.repo.Name, ports.GetJobLogParams{JobID: jobID, Attempt: attempt, Client: s.logClient(r)}, func(b ports.LogBatch) error {
545	550		for _, l := range b.Lines {
546	551		if written > rawLogMaxBytes {
547	552		return errors.New("log too large")

@@ -786,7 +791,7 @@ func (s Server) logStream(w http.ResponseWriter, r http.Request) {
786	791		n := from
787	792		err = s.cfg.Runs.GetJobLog(ctx, principal(ctx), d.repo.Name, ports.GetJobLogParams{
788	793		JobID: jobID, Offset: from, Follow: true,
789		-	Client: clientlimit.Key(clientlimit.ClientAddr(r, s.cfg.TrustedProxies)),
	794	+	Client: s.logClient(r),
790	795		}, func(b ports.LogBatch) error {
791	796		var html strings.Builder
792	797		for i, l := range b.Lines {

 	}
 	p.Follow = !view.Ended()
 	p.StreamHref = jobHref(d.name(), id, jobID) + "/stream"
-	lines, next, err := s.readLog(ctx, d.repo.Name, view, from, jobLogPageLines)
+	lines, next, err := s.readLog(ctx, s.logClient(r), d.repo.Name, view, from, jobLogPageLines)
 	if err != nil {
 		return nil, 0, err
 	}
 	return Job(l, p), http.StatusOK, nil
 }
+// logClient is the key of the visitor for the limit of log streams per client.
+func (s *Server) logClient(r *http.Request) string {
+	return clientlimit.Key(clientlimit.ClientAddr(r, s.cfg.TrustedProxies))
+}
 // readLog reads up to max lines from a line number, through the log stream of
 // the use case, and returns the line after the last read.
-func (s *Server) readLog(ctx context.Context, repo domain.RepoName, j ports.JobView, from int64, max int) ([]LogLineView, int64, error) {
+func (s *Server) readLog(ctx context.Context, client string, repo domain.RepoName, j ports.JobView, from int64, max int) ([]LogLineView, int64, error) {
 	if from > j.Lines || j.Lines == 0 {
 		return nil, from, nil
 	}
 		return 0
 	}
 	stop := errors.New("enough")
-	err := s.cfg.Runs.GetJobLog(ctx, principal(ctx), repo, ports.GetJobLogParams{JobID: j.ID, Attempt: j.Attempt, Offset: from}, func(b ports.LogBatch) error {
+	err := s.cfg.Runs.GetJobLog(ctx, principal(ctx), repo, ports.GetJobLogParams{JobID: j.ID, Attempt: j.Attempt, Offset: from, Client: client}, func(b ports.LogBatch) error {
 		for i, l := range b.Lines {
 			if len(out) >= max {
 				return stop
 	cacheHeaders(ctx, h, "no-cache", sessionCache)
 	written := 0
 	open := false
-	err = s.cfg.Runs.GetJobLog(ctx, principal(ctx), d.repo.Name, ports.GetJobLogParams{JobID: jobID, Attempt: attempt}, func(b ports.LogBatch) error {
+	err = s.cfg.Runs.GetJobLog(ctx, principal(ctx), d.repo.Name, ports.GetJobLogParams{JobID: jobID, Attempt: attempt, Client: s.logClient(r)}, func(b ports.LogBatch) error {
 		for _, l := range b.Lines {
 			if written > rawLogMaxBytes {
 				return errors.New("log too large")
 	n := from
 	err = s.cfg.Runs.GetJobLog(ctx, principal(ctx), d.repo.Name, ports.GetJobLogParams{
 		JobID: jobID, Offset: from, Follow: true,
-		Client: clientlimit.Key(clientlimit.ClientAddr(r, s.cfg.TrustedProxies)),
+		Client: s.logClient(r),
 	}, func(b ports.LogBatch) error {
 		var html strings.Builder
 		for i, l := range b.Lines {

A internal/adapters/web/actions_test.go +70 -0

@@ -0,0 +1,70 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package web
	4	+
	5	+	import (
	6	+	"context"
	7	+	"net/http"
	8	+	"net/http/httptest"
	9	+	"net/netip"
	10	+	"testing"
	11	+	"time"
	12	+
	13	+	"github.com/stretchr/testify/assert"
	14	+	"github.com/stretchr/testify/mock"
	15	+	"github.com/stretchr/testify/require"
	16	+
	17	+	"bnema.dev/ermite/internal/clientlimit"
	18	+	"bnema.dev/ermite/internal/domain"
	19	+	"bnema.dev/ermite/internal/ports"
	20	+	"bnema.dev/ermite/internal/ports/mocks"
	21	+	)
	22	+
	23	+	// S6: every page that streams a log through the use case names the visitor, so
	24	+	// that the limit of log streams per client applies to the pages too, behind a
	25	+	// trusted proxy by the address it forwards.
	26	+	func TestLogPagesPassTheClientKey(t *testing.T) {
	27	+	t.Parallel()
	28	+	const jobID = "1-0-0"
	29	+	repo := repoName(t, "proj")
	30	+	job := ports.JobView{CIJob: domain.CIJob{
	31	+	ID: jobID, RunID: 1, Attempt: 1, RunAttempt: 1, Repo: repo, Key: "build", Name: "build",
	32	+	Phase: domain.CIPhaseEnded, Conclusion: domain.CISuccess,
	33	+	}, Lines: 1}
	34	+	want := clientlimit.Key("198.51.100.7")
	35	+
	36	+	runs := mocks.NewRunService(t)
	37	+	runs.EXPECT().GetJob(mock.Anything, mock.Anything, repo, jobID, int32(0)).Return(job, nil).Maybe()
	38	+	runs.EXPECT().GetRun(mock.Anything, mock.Anything, repo, int64(1), int32(0)).Return(ports.RunDetail{
	39	+	Run: domain.CIRun{Repo: repo, ID: 1, Attempt: 1}, LatestAttempt: 1,
	40	+	}, nil).Maybe()
	41	+	var seen []string
	42	+	runs.EXPECT().GetJobLog(mock.Anything, mock.Anything, repo, mock.Anything, mock.Anything).RunAndReturn(
	43	+	func(_ context.Context, _ domain.Principal, _ domain.RepoName, p ports.GetJobLogParams, send func(ports.LogBatch) error) error {
	44	+	seen = append(seen, p.Client)
	45	+	return send(ports.LogBatch{Lines: []ports.JobLogLine{{Step: 1, CILogLine: domain.CILogLine{Time: time.Unix(0, 0), Text: "x"}}}})
	46	+	})
	47	+
	48	+	browse := mocks.NewBrowseService(t)
	49	+	browse.EXPECT().ListRefs(mock.Anything, mock.Anything, repo, mock.Anything).Return(ports.RefPage{}, nil).Maybe()
	50	+	srv := newTestServer(t, Config{
	51	+	Repos: reposServing(t, publicRepo(t, "proj")), Browse: browse, Runs: runs,
	52	+	TrustedProxies: []netip.Prefix{netip.MustParsePrefix("192.0.2.0/24")},
	53	+	})
	54	+	for _, path := range []string{
	55	+	"/proj/actions/runs/1/job/" + jobID,
	56	+	"/proj/actions/runs/1/job/" + jobID + "/raw",
	57	+	"/proj/actions/runs/1/job/" + jobID + "/stream",
	58	+	} {
	59	+	req := httptest.NewRequest(http.MethodGet, path, nil)
	60	+	req.RemoteAddr = "192.0.2.1:4000"
	61	+	req.Header.Set("X-Forwarded-For", "198.51.100.7")
	62	+	rec := httptest.NewRecorder()
	63	+	srv.ServeHTTP(rec, req)
	64	+	require.Equal(t, http.StatusOK, rec.Code, path)
	65	+	}
	66	+	require.Len(t, seen, 3)
	67	+	for _, c := range seen {
	68	+	assert.Equal(t, want, c)
	69	+	}
	70	+	}

@@ -0,0 +1,70 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package web
	4	+
	5	+	import (
	6	+	"context"
	7	+	"net/http"
	8	+	"net/http/httptest"
	9	+	"net/netip"
	10	+	"testing"
	11	+	"time"
	12	+
	13	+	"github.com/stretchr/testify/assert"
	14	+	"github.com/stretchr/testify/mock"
	15	+	"github.com/stretchr/testify/require"
	16	+
	17	+	"bnema.dev/ermite/internal/clientlimit"
	18	+	"bnema.dev/ermite/internal/domain"
	19	+	"bnema.dev/ermite/internal/ports"
	20	+	"bnema.dev/ermite/internal/ports/mocks"
	21	+	)
	22	+
	23	+	// S6: every page that streams a log through the use case names the visitor, so
	24	+	// that the limit of log streams per client applies to the pages too, behind a
	25	+	// trusted proxy by the address it forwards.
	26	+	func TestLogPagesPassTheClientKey(t *testing.T) {
	27	+	t.Parallel()
	28	+	const jobID = "1-0-0"
	29	+	repo := repoName(t, "proj")
	30	+	job := ports.JobView{CIJob: domain.CIJob{
	31	+	ID: jobID, RunID: 1, Attempt: 1, RunAttempt: 1, Repo: repo, Key: "build", Name: "build",
	32	+	Phase: domain.CIPhaseEnded, Conclusion: domain.CISuccess,
	33	+	}, Lines: 1}
	34	+	want := clientlimit.Key("198.51.100.7")
	35	+
	36	+	runs := mocks.NewRunService(t)
	37	+	runs.EXPECT().GetJob(mock.Anything, mock.Anything, repo, jobID, int32(0)).Return(job, nil).Maybe()
	38	+	runs.EXPECT().GetRun(mock.Anything, mock.Anything, repo, int64(1), int32(0)).Return(ports.RunDetail{
	39	+	Run: domain.CIRun{Repo: repo, ID: 1, Attempt: 1}, LatestAttempt: 1,
	40	+	}, nil).Maybe()
	41	+	var seen []string
	42	+	runs.EXPECT().GetJobLog(mock.Anything, mock.Anything, repo, mock.Anything, mock.Anything).RunAndReturn(
	43	+	func(_ context.Context, _ domain.Principal, _ domain.RepoName, p ports.GetJobLogParams, send func(ports.LogBatch) error) error {
	44	+	seen = append(seen, p.Client)
	45	+	return send(ports.LogBatch{Lines: []ports.JobLogLine{{Step: 1, CILogLine: domain.CILogLine{Time: time.Unix(0, 0), Text: "x"}}}})
	46	+	})
	47	+
	48	+	browse := mocks.NewBrowseService(t)
	49	+	browse.EXPECT().ListRefs(mock.Anything, mock.Anything, repo, mock.Anything).Return(ports.RefPage{}, nil).Maybe()
	50	+	srv := newTestServer(t, Config{
	51	+	Repos: reposServing(t, publicRepo(t, "proj")), Browse: browse, Runs: runs,
	52	+	TrustedProxies: []netip.Prefix{netip.MustParsePrefix("192.0.2.0/24")},
	53	+	})
	54	+	for _, path := range []string{
	55	+	"/proj/actions/runs/1/job/" + jobID,
	56	+	"/proj/actions/runs/1/job/" + jobID + "/raw",
	57	+	"/proj/actions/runs/1/job/" + jobID + "/stream",
	58	+	} {
	59	+	req := httptest.NewRequest(http.MethodGet, path, nil)
	60	+	req.RemoteAddr = "192.0.2.1:4000"
	61	+	req.Header.Set("X-Forwarded-For", "198.51.100.7")
	62	+	rec := httptest.NewRecorder()
	63	+	srv.ServeHTTP(rec, req)
	64	+	require.Equal(t, http.StatusOK, rec.Code, path)
	65	+	}
	66	+	require.Len(t, seen, 3)
	67	+	for _, c := range seen {
	68	+	assert.Equal(t, want, c)
	69	+	}
	70	+	}