refactor(rpc): flush Watch headers with a nil send and mount handlers directly

commit: 1b886226a3f78176e31a54b4326f669809a3db89
author: brice · 12d ago · 2026-10-10 10:38 UTC
parent: 1930d9b
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +19 -60

M internal/adapters/rpc/api.go +6 -48


@@ -53,7 +53,6 @@ type Config struct {
53	53
54	54		// API is the set of Connect handlers of the contract.
55	55		type API struct {
56		-	cfg Config
57	56		handler http.Handler
58	57		}
59	58

@@ -62,8 +61,6 @@ func New(cfg Config) *API {
62	61		if cfg.Limiter == nil {
63	62		cfg.Limiter = clientlimit.NewLimiter(nil)
64	63		}
65		-	a := &API{cfg: cfg}
66		-
67	64		opts := []connect.HandlerOption{
68	65		connect.WithInterceptors(newInterceptor(cfg)),
69	66		connect.WithReadMaxBytes(MaxRequestBytes),

@@ -74,28 +71,12 @@ func New(cfg Config) *API {
74	71		}),
75	72		}
76	73		mux := http.NewServeMux()
77		-	for _, mount := range []func() (string, http.Handler){
78		-	func() (string, http.Handler) {
79		-	return ermiteconnect.NewRepoServiceHandler(&repoService{cfg: cfg}, opts...)
80		-	},
81		-	func() (string, http.Handler) {
82		-	return ermiteconnect.NewGitServiceHandler(&gitService{browse: cfg.Browse}, opts...)
83		-	},
84		-	func() (string, http.Handler) {
85		-	return ermiteconnect.NewTokenServiceHandler(&tokenService{tokens: cfg.Tokens}, opts...)
86		-	},
87		-	func() (string, http.Handler) {
88		-	return ermiteconnect.NewEventServiceHandler(&eventService{events: cfg.Events}, opts...)
89		-	},
90		-	func() (string, http.Handler) {
91		-	return ermiteconnect.NewServerServiceHandler(&serverService{version: cfg.Version}, opts...)
92		-	},
93		-	} {
94		-	path, h := mount()
95		-	mux.Handle(path, h)
96		-	}
97		-	a.handler = withStreamStart(mux)
98		-	return a
	74	+	mux.Handle(ermiteconnect.NewRepoServiceHandler(&repoService{cfg.Repos}, opts...))
	75	+	mux.Handle(ermiteconnect.NewGitServiceHandler(&gitService{cfg.Browse}, opts...))
	76	+	mux.Handle(ermiteconnect.NewTokenServiceHandler(&tokenService{cfg.Tokens}, opts...))
	77	+	mux.Handle(ermiteconnect.NewEventServiceHandler(&eventService{cfg.Events}, opts...))
	78	+	mux.Handle(ermiteconnect.NewServerServiceHandler(&serverService{cfg.Version}, opts...))
	79	+	return &API{handler: mux}
99	80		}
100	81
101	82		// Handler returns the handler of all services. Its paths are the procedure

@@ -109,26 +90,3 @@ func caller(ctx context.Context) domain.Principal {
109	90		p, _ := domain.PrincipalFromContext(ctx)
110	91		return p
111	92		}
112		-
113		-	type startedKey struct{}
114		-
115		-	// withStreamStart gives the handlers a way to send the response headers
116		-	// before the first message of a stream (see startedFunc).
117		-	func withStreamStart(next http.Handler) http.Handler {
118		-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
119		-	start := func() {
120		-	w.WriteHeader(http.StatusOK)
121		-	_ = http.NewResponseController(w).Flush()
122		-	}
123		-	next.ServeHTTP(w, r.WithContext(context.WithValue(r.Context(), startedKey{}, start)))
124		-	})
125		-	}
126		-
127		-	// startedFunc returns the function that sends the response headers at once.
128		-	// A Connect client returns from a streaming call only when the headers have
129		-	// arrived; a stream that may stay silent for a long time (Watch) must send
130		-	// them first. Connect has set all the headers by the time a handler runs.
131		-	func startedFunc(ctx context.Context) func() {
132		-	start, _ := ctx.Value(startedKey{}).(func())
133		-	return start
134		-	}


@@ -53,7 +53,6 @@ type Config struct {
53			53
54		// API is the set of Connect handlers of the contract.	54		// API is the set of Connect handlers of the contract.
55		type API struct {	55		type API struct {
56	-	cfg Config
57		handler http.Handler	56		handler http.Handler
58		}	57		}
59			58

@@ -62,8 +61,6 @@ func New(cfg Config) *API {
62		if cfg.Limiter == nil {	61		if cfg.Limiter == nil {
63		cfg.Limiter = clientlimit.NewLimiter(nil)	62		cfg.Limiter = clientlimit.NewLimiter(nil)
64		}	63		}
65	-	a := &API{cfg: cfg}
66	-
67		opts := []connect.HandlerOption{	64		opts := []connect.HandlerOption{
68		connect.WithInterceptors(newInterceptor(cfg)),	65		connect.WithInterceptors(newInterceptor(cfg)),
69		connect.WithReadMaxBytes(MaxRequestBytes),	66		connect.WithReadMaxBytes(MaxRequestBytes),

@@ -74,28 +71,12 @@ func New(cfg Config) *API {
74		}),	71		}),
75		}	72		}
76		mux := http.NewServeMux()	73		mux := http.NewServeMux()
77	-	for _, mount := range []func() (string, http.Handler){	74	+	mux.Handle(ermiteconnect.NewRepoServiceHandler(&repoService{cfg.Repos}, opts...))
78	-	func() (string, http.Handler) {	75	+	mux.Handle(ermiteconnect.NewGitServiceHandler(&gitService{cfg.Browse}, opts...))
79	-	return ermiteconnect.NewRepoServiceHandler(&repoService{cfg: cfg}, opts...)	76	+	mux.Handle(ermiteconnect.NewTokenServiceHandler(&tokenService{cfg.Tokens}, opts...))
80	-	},	77	+	mux.Handle(ermiteconnect.NewEventServiceHandler(&eventService{cfg.Events}, opts...))
81	-	func() (string, http.Handler) {	78	+	mux.Handle(ermiteconnect.NewServerServiceHandler(&serverService{cfg.Version}, opts...))
82	-	return ermiteconnect.NewGitServiceHandler(&gitService{browse: cfg.Browse}, opts...)	79	+	return &API{handler: mux}
83	-	},
84	-	func() (string, http.Handler) {
85	-	return ermiteconnect.NewTokenServiceHandler(&tokenService{tokens: cfg.Tokens}, opts...)
86	-	},
87	-	func() (string, http.Handler) {
88	-	return ermiteconnect.NewEventServiceHandler(&eventService{events: cfg.Events}, opts...)
89	-	},
90	-	func() (string, http.Handler) {
91	-	return ermiteconnect.NewServerServiceHandler(&serverService{version: cfg.Version}, opts...)
92	-	},
93	-	} {
94	-	path, h := mount()
95	-	mux.Handle(path, h)
96	-	}
97	-	a.handler = withStreamStart(mux)
98	-	return a
99		}	80		}
100			81
101		// Handler returns the handler of all services. Its paths are the procedure	82		// Handler returns the handler of all services. Its paths are the procedure

@@ -109,26 +90,3 @@ func caller(ctx context.Context) domain.Principal {
109		p, _ := domain.PrincipalFromContext(ctx)	90		p, _ := domain.PrincipalFromContext(ctx)
110		return p	91		return p
111		}	92		}
112	-
113	-	type startedKey struct{}
114	-
115	-	// withStreamStart gives the handlers a way to send the response headers
116	-	// before the first message of a stream (see startedFunc).
117	-	func withStreamStart(next http.Handler) http.Handler {
118	-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
119	-	start := func() {
120	-	w.WriteHeader(http.StatusOK)
121	-	_ = http.NewResponseController(w).Flush()
122	-	}
123	-	next.ServeHTTP(w, r.WithContext(context.WithValue(r.Context(), startedKey{}, start)))
124	-	})
125	-	}
126	-
127	-	// startedFunc returns the function that sends the response headers at once.
128	-	// A Connect client returns from a streaming call only when the headers have
129	-	// arrived; a stream that may stay silent for a long time (Watch) must send
130	-	// them first. Connect has set all the headers by the time a handler runs.
131	-	func startedFunc(ctx context.Context) func() {
132	-	start, _ := ctx.Value(startedKey{}).(func())
133	-	return start
134	-	}

M internal/adapters/rpc/event.go +3 -2


@@ -21,8 +21,9 @@ func (s eventService) Watch(ctx context.Context, req connect.Request[pb.WatchR
21	21		params := app.WatchParams{
22	22		Topics: req.Msg.GetTopics(), ResumeAfter: req.Msg.GetResumeAfter(),
23	23		// A client's call returns when the response begins: send the headers
24		-	// now, not with the first event, which may be far away.
25		-	Ready: startedFunc(ctx),
	24	+	// now, not with the first event, which may be far away. Connect flushes
	25	+	// them for a nil message. A failure shows on the first real send.
	26	+	Ready: func() { _ = stream.Send(nil) },
26	27		}
27	28		err := s.events.Watch(ctx, caller(ctx), params, func(we app.WatchEvent) error {
28	29		ev := watchEventToPB(we)

 	params := app.WatchParams{
 		Topics: req.Msg.GetTopics(), ResumeAfter: req.Msg.GetResumeAfter(),
 		// A client's call returns when the response begins: send the headers
-		// now, not with the first event, which may be far away.
+		// now, not with the first event, which may be far away. Connect flushes
-		Ready: startedFunc(ctx),
+		// them for a nil message. A failure shows on the first real send.
+		Ready: func() { _ = stream.Send(nil) },
 	}
 	err := s.events.Watch(ctx, caller(ctx), params, func(we app.WatchEvent) error {
 		ev := watchEventToPB(we)

M internal/adapters/rpc/repo.go +10 -10


@@ -14,16 +14,16 @@ import (
14	14		)
15	15
16	16		// repoService implements ermiteconnect.RepoServiceHandler.
17		-	type repoService struct{ cfg Config }
	17	+	type repoService struct{ repos *app.Repos }
18	18
19	19		func (s repoService) ListRepos(ctx context.Context, req connect.Request[pb.ListReposRequest]) (*connect.Response[pb.ListReposResponse], error) {
20		-	page, err := s.cfg.Repos.ListPage(ctx, caller(ctx), int(req.Msg.GetPageSize()), req.Msg.GetPageToken())
	20	+	page, err := s.repos.ListPage(ctx, caller(ctx), int(req.Msg.GetPageSize()), req.Msg.GetPageToken())
21	21		if err != nil {
22	22		return nil, connectError(ctx, err)
23	23		}
24	24		out := &pb.ListReposResponse{NextPageToken: page.NextPageToken, Repos: make([]*pb.Repo, 0, len(page.Repos))}
25	25		for _, repo := range page.Repos {
26		-	view, err := s.cfg.Repos.View(ctx, repo)
	26	+	view, err := s.repos.View(ctx, repo)
27	27		if err != nil {
28	28		return nil, connectError(ctx, err)
29	29		}

@@ -37,11 +37,11 @@ func (s repoService) GetRepo(ctx context.Context, req connect.Request[pb.GetRe
37	37		if err != nil {
38	38		return nil, connectError(ctx, err)
39	39		}
40		-	repo, err := s.cfg.Repos.Get(ctx, caller(ctx), name)
	40	+	repo, err := s.repos.Get(ctx, caller(ctx), name)
41	41		if err != nil {
42	42		return nil, connectError(ctx, err)
43	43		}
44		-	return respondRepo(ctx, s.cfg.Repos, repo, func(r pb.Repo) pb.GetRepoResponse { return &pb.GetRepoResponse{Repo: r} })
	44	+	return respondRepo(ctx, s.repos, repo, func(r pb.Repo) pb.GetRepoResponse { return &pb.GetRepoResponse{Repo: r} })
45	45		}
46	46
47	47		func (s repoService) CreateRepo(ctx context.Context, req connect.Request[pb.CreateRepoRequest]) (*connect.Response[pb.CreateRepoResponse], error) {

@@ -53,13 +53,13 @@ func (s repoService) CreateRepo(ctx context.Context, req connect.Request[pb.Cr
53	53		if err != nil {
54	54		return nil, connectError(ctx, err)
55	55		}
56		-	repo, err := s.cfg.Repos.Create(ctx, caller(ctx), app.CreateParams{
	56	+	repo, err := s.repos.Create(ctx, caller(ctx), app.CreateParams{
57	57		Name: name, Description: req.Msg.GetDescription(), Visibility: vis,
58	58		})
59	59		if err != nil {
60	60		return nil, connectError(ctx, err)
61	61		}
62		-	return respondRepo(ctx, s.cfg.Repos, repo, func(r pb.Repo) pb.CreateRepoResponse { return &pb.CreateRepoResponse{Repo: r} })
	62	+	return respondRepo(ctx, s.repos, repo, func(r pb.Repo) pb.CreateRepoResponse { return &pb.CreateRepoResponse{Repo: r} })
63	63		}
64	64
65	65		func (s repoService) UpdateRepo(ctx context.Context, req connect.Request[pb.UpdateRepoRequest]) (*connect.Response[pb.UpdateRepoResponse], error) {

@@ -71,11 +71,11 @@ func (s repoService) UpdateRepo(ctx context.Context, req connect.Request[pb.Up
71	71		if err != nil {
72	72		return nil, connectError(ctx, err)
73	73		}
74		-	repo, err := s.cfg.Repos.Update(ctx, caller(ctx), name, params)
	74	+	repo, err := s.repos.Update(ctx, caller(ctx), name, params)
75	75		if err != nil {
76	76		return nil, connectError(ctx, err)
77	77		}
78		-	return respondRepo(ctx, s.cfg.Repos, repo, func(r pb.Repo) pb.UpdateRepoResponse { return &pb.UpdateRepoResponse{Repo: r} })
	78	+	return respondRepo(ctx, s.repos, repo, func(r pb.Repo) pb.UpdateRepoResponse { return &pb.UpdateRepoResponse{Repo: r} })
79	79		}
80	80
81	81		// updateParams maps the field mask of an update. The mask must name at least

@@ -115,7 +115,7 @@ func (s repoService) DeleteRepo(ctx context.Context, req connect.Request[pb.De
115	115		if err != nil {
116	116		return nil, connectError(ctx, err)
117	117		}
118		-	if err := s.cfg.Repos.Delete(ctx, caller(ctx), name); err != nil {
	118	+	if err := s.repos.Delete(ctx, caller(ctx), name); err != nil {
119	119		return nil, connectError(ctx, err)
120	120		}
121	121		return connect.NewResponse(&pb.DeleteRepoResponse{}), nil


@@ -14,16 +14,16 @@ import (
14		)	14		)
15			15
16		// repoService implements ermiteconnect.RepoServiceHandler.	16		// repoService implements ermiteconnect.RepoServiceHandler.
17	-	type repoService struct{ cfg Config }	17	+	type repoService struct{ repos *app.Repos }
18			18
19		func (s repoService) ListRepos(ctx context.Context, req connect.Request[pb.ListReposRequest]) (*connect.Response[pb.ListReposResponse], error) {	19		func (s repoService) ListRepos(ctx context.Context, req connect.Request[pb.ListReposRequest]) (*connect.Response[pb.ListReposResponse], error) {
20	-	page, err := s.cfg.Repos.ListPage(ctx, caller(ctx), int(req.Msg.GetPageSize()), req.Msg.GetPageToken())	20	+	page, err := s.repos.ListPage(ctx, caller(ctx), int(req.Msg.GetPageSize()), req.Msg.GetPageToken())
21		if err != nil {	21		if err != nil {
22		return nil, connectError(ctx, err)	22		return nil, connectError(ctx, err)
23		}	23		}
24		out := &pb.ListReposResponse{NextPageToken: page.NextPageToken, Repos: make([]*pb.Repo, 0, len(page.Repos))}	24		out := &pb.ListReposResponse{NextPageToken: page.NextPageToken, Repos: make([]*pb.Repo, 0, len(page.Repos))}
25		for _, repo := range page.Repos {	25		for _, repo := range page.Repos {
26	-	view, err := s.cfg.Repos.View(ctx, repo)	26	+	view, err := s.repos.View(ctx, repo)
27		if err != nil {	27		if err != nil {
28		return nil, connectError(ctx, err)	28		return nil, connectError(ctx, err)
29		}	29		}

@@ -37,11 +37,11 @@ func (s repoService) GetRepo(ctx context.Context, req connect.Request[pb.GetRe
37		if err != nil {	37		if err != nil {
38		return nil, connectError(ctx, err)	38		return nil, connectError(ctx, err)
39		}	39		}
40	-	repo, err := s.cfg.Repos.Get(ctx, caller(ctx), name)	40	+	repo, err := s.repos.Get(ctx, caller(ctx), name)
41		if err != nil {	41		if err != nil {
42		return nil, connectError(ctx, err)	42		return nil, connectError(ctx, err)
43		}	43		}
44	-	return respondRepo(ctx, s.cfg.Repos, repo, func(r pb.Repo) pb.GetRepoResponse { return &pb.GetRepoResponse{Repo: r} })	44	+	return respondRepo(ctx, s.repos, repo, func(r pb.Repo) pb.GetRepoResponse { return &pb.GetRepoResponse{Repo: r} })
45		}	45		}
46			46
47		func (s repoService) CreateRepo(ctx context.Context, req connect.Request[pb.CreateRepoRequest]) (*connect.Response[pb.CreateRepoResponse], error) {	47		func (s repoService) CreateRepo(ctx context.Context, req connect.Request[pb.CreateRepoRequest]) (*connect.Response[pb.CreateRepoResponse], error) {

@@ -53,13 +53,13 @@ func (s repoService) CreateRepo(ctx context.Context, req connect.Request[pb.Cr
53		if err != nil {	53		if err != nil {
54		return nil, connectError(ctx, err)	54		return nil, connectError(ctx, err)
55		}	55		}
56	-	repo, err := s.cfg.Repos.Create(ctx, caller(ctx), app.CreateParams{	56	+	repo, err := s.repos.Create(ctx, caller(ctx), app.CreateParams{
57		Name: name, Description: req.Msg.GetDescription(), Visibility: vis,	57		Name: name, Description: req.Msg.GetDescription(), Visibility: vis,
58		})	58		})
59		if err != nil {	59		if err != nil {
60		return nil, connectError(ctx, err)	60		return nil, connectError(ctx, err)
61		}	61		}
62	-	return respondRepo(ctx, s.cfg.Repos, repo, func(r pb.Repo) pb.CreateRepoResponse { return &pb.CreateRepoResponse{Repo: r} })	62	+	return respondRepo(ctx, s.repos, repo, func(r pb.Repo) pb.CreateRepoResponse { return &pb.CreateRepoResponse{Repo: r} })
63		}	63		}
64			64
65		func (s repoService) UpdateRepo(ctx context.Context, req connect.Request[pb.UpdateRepoRequest]) (*connect.Response[pb.UpdateRepoResponse], error) {	65		func (s repoService) UpdateRepo(ctx context.Context, req connect.Request[pb.UpdateRepoRequest]) (*connect.Response[pb.UpdateRepoResponse], error) {

@@ -71,11 +71,11 @@ func (s repoService) UpdateRepo(ctx context.Context, req connect.Request[pb.Up
71		if err != nil {	71		if err != nil {
72		return nil, connectError(ctx, err)	72		return nil, connectError(ctx, err)
73		}	73		}
74	-	repo, err := s.cfg.Repos.Update(ctx, caller(ctx), name, params)	74	+	repo, err := s.repos.Update(ctx, caller(ctx), name, params)
75		if err != nil {	75		if err != nil {
76		return nil, connectError(ctx, err)	76		return nil, connectError(ctx, err)
77		}	77		}
78	-	return respondRepo(ctx, s.cfg.Repos, repo, func(r pb.Repo) pb.UpdateRepoResponse { return &pb.UpdateRepoResponse{Repo: r} })	78	+	return respondRepo(ctx, s.repos, repo, func(r pb.Repo) pb.UpdateRepoResponse { return &pb.UpdateRepoResponse{Repo: r} })
79		}	79		}
80			80
81		// updateParams maps the field mask of an update. The mask must name at least	81		// updateParams maps the field mask of an update. The mask must name at least

@@ -115,7 +115,7 @@ func (s repoService) DeleteRepo(ctx context.Context, req connect.Request[pb.De
115		if err != nil {	115		if err != nil {
116		return nil, connectError(ctx, err)	116		return nil, connectError(ctx, err)
117		}	117		}
118	-	if err := s.cfg.Repos.Delete(ctx, caller(ctx), name); err != nil {	118	+	if err := s.repos.Delete(ctx, caller(ctx), name); err != nil {
119		return nil, connectError(ctx, err)	119		return nil, connectError(ctx, err)
120		}	120		}
121		return connect.NewResponse(&pb.DeleteRepoResponse{}), nil	121		return connect.NewResponse(&pb.DeleteRepoResponse{}), nil