feat(serve): wire push use cases, event bus and debounced maintenance

commit: 979aa64b8fae9093d2862757474eedbf90da5552
author: brice · 12d ago · 2026-10-10 07:10 UTC
parent: d949cad
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

7 files · +721 -47

M cmd/ermite/datadir.go +6 -4


@@ -18,16 +18,18 @@ func prepareDataDir(root string) error {
18	18		return nil
19	19		}
20	20
21		-	// resetRunDir recreates run/ and run/git-home/, which hold nothing that
22		-	// survives a start. Call it only once the kv directory lock is held: a second
	21	+	// resetRunDir recreates run/ with git-home/ and tmp/, which hold nothing
	22	+	// that survives a start. Call it only once the kv directory lock is held: a second
23	23		// instance on the same data root must fail before touching run/.
24	24		func resetRunDir(root string) error {
25	25		run := filepath.Join(root, "run")
26	26		if err := os.RemoveAll(run); err != nil {
27	27		return fmt.Errorf("reset %s: %w", run, err)
28	28		}
29		-	if err := os.MkdirAll(filepath.Join(run, "git-home"), 0o700); err != nil {
30		-	return fmt.Errorf("prepare %s: %w", run, err)
	29	+	for _, d := range []string{"git-home", "tmp"} {
	30	+	if err := os.MkdirAll(filepath.Join(run, d), 0o700); err != nil {
	31	+	return fmt.Errorf("prepare %s: %w", run, err)
	32	+	}
31	33		}
32	34		return nil
33	35		}

 	return nil
 }
-// resetRunDir recreates run/ and run/git-home/, which hold nothing that
+// resetRunDir recreates run/ with git-home/ and tmp/, which hold nothing
-// survives a start. Call it only once the kv directory lock is held: a second
+// that survives a start. Call it only once the kv directory lock is held: a second
 // instance on the same data root must fail before touching run/.
 func resetRunDir(root string) error {
 	run := filepath.Join(root, "run")
 	if err := os.RemoveAll(run); err != nil {
 		return fmt.Errorf("reset %s: %w", run, err)
 	}
-	if err := os.MkdirAll(filepath.Join(run, "git-home"), 0o700); err != nil {
+	for _, d := range []string{"git-home", "tmp"} {
-		return fmt.Errorf("prepare %s: %w", run, err)
+		if err := os.MkdirAll(filepath.Join(run, d), 0o700); err != nil {
+			return fmt.Errorf("prepare %s: %w", run, err)
+		}
 	}
 	return nil
 }

A cmd/ermite/forge.go +186 -0

@@ -0,0 +1,186 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package main
	4	+
	5	+	import (
	6	+	"context"
	7	+	"errors"
	8	+	"fmt"
	9	+	"net"
	10	+	"os"
	11	+	"path/filepath"
	12	+	"time"
	13	+
	14	+	"github.com/bnema/zerowrap"
	15	+
	16	+	"bnema.dev/ermite/internal/adapters/eventbus"
	17	+	"bnema.dev/ermite/internal/adapters/gitcli"
	18	+	"bnema.dev/ermite/internal/adapters/githttp"
	19	+	"bnema.dev/ermite/internal/adapters/httpserver"
	20	+	"bnema.dev/ermite/internal/adapters/kvstore"
	21	+	"bnema.dev/ermite/internal/app"
	22	+	"bnema.dev/ermite/internal/config"
	23	+	"bnema.dev/ermite/internal/ports"
	24	+	)
	25	+
	26	+	// options are the parts of the wiring that tests replace.
	27	+	type options struct {
	28	+	// hookExe is the executable the hook scripts exec; empty means this
	29	+	// process's own executable.
	30	+	hookExe string
	31	+	// maintenanceDelay overrides app.MaintenanceDelay when positive.
	32	+	maintenanceDelay time.Duration
	33	+	}
	34	+
	35	+	// forge is the wired server: every adapter and use case, ready to serve.
	36	+	type forge struct {
	37	+	cfg config.Config
	38	+	store *kvstore.Store
	39	+	server *httpserver.Server
	40	+	hooks *githttp.HookServer
	41	+	maint *app.Maintainer
	42	+
	43	+	// Repos, Tokens, Pushes and Events are exposed for tests and for the
	44	+	// phases that add the API.
	45	+	Repos *app.Repos
	46	+	Tokens *app.Tokens
	47	+	Auth *app.Auth
	48	+	Pushes *app.Pushes
	49	+	Events ports.EventBus
	50	+	}
	51	+
	52	+	// newForge opens the metadata store, resets run/, checks git, reconciles the repositories,
	53	+	// writes the hook scripts, opens the hook socket and builds the HTTP server.
	54	+	// The data directory must be prepared (prepareDataDir). Close it after the
	55	+	// server stopped.
	56	+	func newForge(ctx context.Context, cfg config.Config, opts options) (_ *forge, err error) {
	57	+	log := zerowrap.FromCtx(ctx)
	58	+	store, err := kvstore.Open(ctx, filepath.Join(cfg.DataDir, "kv"))
	59	+	if err != nil {
	60	+	return nil, err
	61	+	}
	62	+	f := &forge{cfg: cfg, store: store}
	63	+	defer func() {
	64	+	if err != nil {
	65	+	f.Close(context.WithoutCancel(ctx))
	66	+	}
	67	+	}()
	68	+	// The kv store holds the directory lock: only now is run/ ours to reset.
	69	+	// The hook scripts and the socket are written into it afterwards.
	70	+	if err := resetRunDir(cfg.DataDir); err != nil {
	71	+	return nil, err
	72	+	}
	73	+
	74	+	path := os.Getenv("PATH")
	75	+	if path == "" {
	76	+	path = "/usr/local/bin:/usr/bin:/bin"
	77	+	}
	78	+	git := gitcli.New(gitcli.Config{Bin: cfg.Git, DataDir: cfg.DataDir, Path: path})
	79	+	gitVersion, err := git.CheckVersion(ctx)
	80	+	if err != nil {
	81	+	return nil, fmt.Errorf("%s=%q: %w", config.EnvGit, cfg.Git, err)
	82	+	}
	83	+	log.Info().Str("git", gitVersion.String()).Msg("git checked")
	84	+	if _, err := git.HTTPBackend(ctx, false); err != nil {
	85	+	return nil, fmt.Errorf("%s=%q: %w", config.EnvGit, cfg.Git, err)
	86	+	}
	87	+
	88	+	clock := systemClock{}
	89	+	f.Repos = app.NewRepos(store, git, clock)
	90	+	report, err := f.Repos.Reconcile(ctx)
	91	+	if err != nil {
	92	+	return nil, fmt.Errorf("reconcile repositories: %w", err)
	93	+	}
	94	+	log.Info().
	95	+	Int("registered", len(report.Registered)).
	96	+	Int("missing_dirs", len(report.MissingDirs)).
	97	+	Int("invalid_dirs", len(report.InvalidDirs)).
	98	+	Msg("repositories reconciled")
	99	+
	100	+	f.Tokens = app.NewTokens(store.Tokens(), clock)
	101	+	f.Auth = app.NewAuth(cfg.AdminTokenHash, store.Tokens(), clock)
	102	+	f.Events = eventbus.New()
	103	+	f.maint = app.NewMaintainer(ctx, git, opts.maintenanceDelay)
	104	+	f.Pushes = app.NewPushes(store, f.Repos, git, f.Events, clock, f.maint)
	105	+
	106	+	exe := opts.hookExe
	107	+	if exe == "" {
	108	+	if exe, err = os.Executable(); err != nil {
	109	+	return nil, fmt.Errorf("find the ermite executable for the hooks: %w", err)
	110	+	}
	111	+	}
	112	+	if err := githttp.WriteHookScripts(gitcli.HooksDir(cfg.DataDir), exe); err != nil {
	113	+	return nil, err
	114	+	}
	115	+	f.hooks = githttp.NewHookServer(filepath.Join(cfg.DataDir, "run", "hook.sock"), f.Pushes)
	116	+	if err := f.hooks.Listen(); err != nil {
	117	+	return nil, err
	118	+	}
	119	+
	120	+	gh := githttp.New(githttp.Config{
	121	+	Auth: f.Auth,
	122	+	Repos: f.Repos,
	123	+	Pushes: f.Pushes,
	124	+	Backend: git,
	125	+	Hooks: f.hooks,
	126	+	TrustedProxies: cfg.TrustedProxies,
	127	+	TempDir: filepath.Join(cfg.DataDir, "run", "tmp"),
	128	+	})
	129	+	routes := make([]httpserver.Route, 0, len(githttp.Patterns))
	130	+	for _, p := range githttp.Patterns {
	131	+	routes = append(routes, httpserver.Route{Pattern: p, Handler: gh})
	132	+	}
	133	+	f.server = httpserver.New(httpserver.Config{
	134	+	Addr: cfg.Addr,
	135	+	HealthChecks: []httpserver.Check{store.Check, httpserver.DirWritable(cfg.DataDir)},
	136	+	Routes: routes,
	137	+	})
	138	+	return f, nil
	139	+	}
	140	+
	141	+	// Run listens on the configured address and serves until ctx ends.
	142	+	func (f *forge) Run(ctx context.Context) error {
	143	+	ln, err := net.Listen("tcp", f.cfg.Addr)
	144	+	if err != nil {
	145	+	return fmt.Errorf("listen on %s: %w", f.cfg.Addr, err)
	146	+	}
	147	+	return f.Serve(ctx, ln)
	148	+	}
	149	+
	150	+	// Serve serves HTTP on ln and the hook socket until ctx ends. The hook socket
	151	+	// outlives the HTTP server: requests still in flight while it drains are
	152	+	// pushes whose hooks need it. A failure of either stops both.
	153	+	func (f *forge) Serve(ctx context.Context, ln net.Listener) error {
	154	+	ctx, cancel := context.WithCancel(ctx)
	155	+	defer cancel()
	156	+	hookCtx, stopHooks := context.WithCancel(context.WithoutCancel(ctx))
	157	+	defer stopHooks()
	158	+
	159	+	hookErr := make(chan error, 1)
	160	+	go func() {
	161	+	err := f.hooks.Serve(hookCtx)
	162	+	cancel() // a dead hook socket would reject every push: stop serving
	163	+	hookErr <- err
	164	+	}()
	165	+
	166	+	httpErr := f.server.Serve(ctx, ln)
	167	+	stopHooks()
	168	+	return errors.Join(httpErr, <-hookErr)
	169	+	}
	170	+
	171	+	// Close releases what newForge opened. It is safe to call twice.
	172	+	func (f *forge) Close(ctx context.Context) {
	173	+	log := zerowrap.FromCtx(ctx)
	174	+	if f.hooks != nil {
	175	+	f.hooks.Close()
	176	+	}
	177	+	if f.maint != nil {
	178	+	f.maint.Close(ctx)
	179	+	}
	180	+	if f.store != nil {
	181	+	if err := f.store.Close(ctx); err != nil {
	182	+	log.Error().Err(err).Msg("close metadata store")
	183	+	}
	184	+	f.store = nil
	185	+	}
	186	+	}

@@ -0,0 +1,186 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package main
	4	+
	5	+	import (
	6	+	"context"
	7	+	"errors"
	8	+	"fmt"
	9	+	"net"
	10	+	"os"
	11	+	"path/filepath"
	12	+	"time"
	13	+
	14	+	"github.com/bnema/zerowrap"
	15	+
	16	+	"bnema.dev/ermite/internal/adapters/eventbus"
	17	+	"bnema.dev/ermite/internal/adapters/gitcli"
	18	+	"bnema.dev/ermite/internal/adapters/githttp"
	19	+	"bnema.dev/ermite/internal/adapters/httpserver"
	20	+	"bnema.dev/ermite/internal/adapters/kvstore"
	21	+	"bnema.dev/ermite/internal/app"
	22	+	"bnema.dev/ermite/internal/config"
	23	+	"bnema.dev/ermite/internal/ports"
	24	+	)
	25	+
	26	+	// options are the parts of the wiring that tests replace.
	27	+	type options struct {
	28	+	// hookExe is the executable the hook scripts exec; empty means this
	29	+	// process's own executable.
	30	+	hookExe string
	31	+	// maintenanceDelay overrides app.MaintenanceDelay when positive.
	32	+	maintenanceDelay time.Duration
	33	+	}
	34	+
	35	+	// forge is the wired server: every adapter and use case, ready to serve.
	36	+	type forge struct {
	37	+	cfg config.Config
	38	+	store *kvstore.Store
	39	+	server *httpserver.Server
	40	+	hooks *githttp.HookServer
	41	+	maint *app.Maintainer
	42	+
	43	+	// Repos, Tokens, Pushes and Events are exposed for tests and for the
	44	+	// phases that add the API.
	45	+	Repos *app.Repos
	46	+	Tokens *app.Tokens
	47	+	Auth *app.Auth
	48	+	Pushes *app.Pushes
	49	+	Events ports.EventBus
	50	+	}
	51	+
	52	+	// newForge opens the metadata store, resets run/, checks git, reconciles the repositories,
	53	+	// writes the hook scripts, opens the hook socket and builds the HTTP server.
	54	+	// The data directory must be prepared (prepareDataDir). Close it after the
	55	+	// server stopped.
	56	+	func newForge(ctx context.Context, cfg config.Config, opts options) (_ *forge, err error) {
	57	+	log := zerowrap.FromCtx(ctx)
	58	+	store, err := kvstore.Open(ctx, filepath.Join(cfg.DataDir, "kv"))
	59	+	if err != nil {
	60	+	return nil, err
	61	+	}
	62	+	f := &forge{cfg: cfg, store: store}
	63	+	defer func() {
	64	+	if err != nil {
	65	+	f.Close(context.WithoutCancel(ctx))
	66	+	}
	67	+	}()
	68	+	// The kv store holds the directory lock: only now is run/ ours to reset.
	69	+	// The hook scripts and the socket are written into it afterwards.
	70	+	if err := resetRunDir(cfg.DataDir); err != nil {
	71	+	return nil, err
	72	+	}
	73	+
	74	+	path := os.Getenv("PATH")
	75	+	if path == "" {
	76	+	path = "/usr/local/bin:/usr/bin:/bin"
	77	+	}
	78	+	git := gitcli.New(gitcli.Config{Bin: cfg.Git, DataDir: cfg.DataDir, Path: path})
	79	+	gitVersion, err := git.CheckVersion(ctx)
	80	+	if err != nil {
	81	+	return nil, fmt.Errorf("%s=%q: %w", config.EnvGit, cfg.Git, err)
	82	+	}
	83	+	log.Info().Str("git", gitVersion.String()).Msg("git checked")
	84	+	if _, err := git.HTTPBackend(ctx, false); err != nil {
	85	+	return nil, fmt.Errorf("%s=%q: %w", config.EnvGit, cfg.Git, err)
	86	+	}
	87	+
	88	+	clock := systemClock{}
	89	+	f.Repos = app.NewRepos(store, git, clock)
	90	+	report, err := f.Repos.Reconcile(ctx)
	91	+	if err != nil {
	92	+	return nil, fmt.Errorf("reconcile repositories: %w", err)
	93	+	}
	94	+	log.Info().
	95	+	Int("registered", len(report.Registered)).
	96	+	Int("missing_dirs", len(report.MissingDirs)).
	97	+	Int("invalid_dirs", len(report.InvalidDirs)).
	98	+	Msg("repositories reconciled")
	99	+
	100	+	f.Tokens = app.NewTokens(store.Tokens(), clock)
	101	+	f.Auth = app.NewAuth(cfg.AdminTokenHash, store.Tokens(), clock)
	102	+	f.Events = eventbus.New()
	103	+	f.maint = app.NewMaintainer(ctx, git, opts.maintenanceDelay)
	104	+	f.Pushes = app.NewPushes(store, f.Repos, git, f.Events, clock, f.maint)
	105	+
	106	+	exe := opts.hookExe
	107	+	if exe == "" {
	108	+	if exe, err = os.Executable(); err != nil {
	109	+	return nil, fmt.Errorf("find the ermite executable for the hooks: %w", err)
	110	+	}
	111	+	}
	112	+	if err := githttp.WriteHookScripts(gitcli.HooksDir(cfg.DataDir), exe); err != nil {
	113	+	return nil, err
	114	+	}
	115	+	f.hooks = githttp.NewHookServer(filepath.Join(cfg.DataDir, "run", "hook.sock"), f.Pushes)
	116	+	if err := f.hooks.Listen(); err != nil {
	117	+	return nil, err
	118	+	}
	119	+
	120	+	gh := githttp.New(githttp.Config{
	121	+	Auth: f.Auth,
	122	+	Repos: f.Repos,
	123	+	Pushes: f.Pushes,
	124	+	Backend: git,
	125	+	Hooks: f.hooks,
	126	+	TrustedProxies: cfg.TrustedProxies,
	127	+	TempDir: filepath.Join(cfg.DataDir, "run", "tmp"),
	128	+	})
	129	+	routes := make([]httpserver.Route, 0, len(githttp.Patterns))
	130	+	for _, p := range githttp.Patterns {
	131	+	routes = append(routes, httpserver.Route{Pattern: p, Handler: gh})
	132	+	}
	133	+	f.server = httpserver.New(httpserver.Config{
	134	+	Addr: cfg.Addr,
	135	+	HealthChecks: []httpserver.Check{store.Check, httpserver.DirWritable(cfg.DataDir)},
	136	+	Routes: routes,
	137	+	})
	138	+	return f, nil
	139	+	}
	140	+
	141	+	// Run listens on the configured address and serves until ctx ends.
	142	+	func (f *forge) Run(ctx context.Context) error {
	143	+	ln, err := net.Listen("tcp", f.cfg.Addr)
	144	+	if err != nil {
	145	+	return fmt.Errorf("listen on %s: %w", f.cfg.Addr, err)
	146	+	}
	147	+	return f.Serve(ctx, ln)
	148	+	}
	149	+
	150	+	// Serve serves HTTP on ln and the hook socket until ctx ends. The hook socket
	151	+	// outlives the HTTP server: requests still in flight while it drains are
	152	+	// pushes whose hooks need it. A failure of either stops both.
	153	+	func (f *forge) Serve(ctx context.Context, ln net.Listener) error {
	154	+	ctx, cancel := context.WithCancel(ctx)
	155	+	defer cancel()
	156	+	hookCtx, stopHooks := context.WithCancel(context.WithoutCancel(ctx))
	157	+	defer stopHooks()
	158	+
	159	+	hookErr := make(chan error, 1)
	160	+	go func() {
	161	+	err := f.hooks.Serve(hookCtx)
	162	+	cancel() // a dead hook socket would reject every push: stop serving
	163	+	hookErr <- err
	164	+	}()
	165	+
	166	+	httpErr := f.server.Serve(ctx, ln)
	167	+	stopHooks()
	168	+	return errors.Join(httpErr, <-hookErr)
	169	+	}
	170	+
	171	+	// Close releases what newForge opened. It is safe to call twice.
	172	+	func (f *forge) Close(ctx context.Context) {
	173	+	log := zerowrap.FromCtx(ctx)
	174	+	if f.hooks != nil {
	175	+	f.hooks.Close()
	176	+	}
	177	+	if f.maint != nil {
	178	+	f.maint.Close(ctx)
	179	+	}
	180	+	if f.store != nil {
	181	+	if err := f.store.Close(ctx); err != nil {
	182	+	log.Error().Err(err).Msg("close metadata store")
	183	+	}
	184	+	f.store = nil
	185	+	}
	186	+	}

M cmd/ermite/serve.go +3 -43


@@ -4,19 +4,13 @@ package main
4	4
5	5		import (
6	6		"context"
7		-	"fmt"
8	7		"os"
9	8		"os/signal"
10		-	"path/filepath"
11	9		"syscall"
12	10
13	11		"github.com/bnema/zerowrap"
14	12		"github.com/spf13/cobra"
15	13
16		-	"bnema.dev/ermite/internal/adapters/gitcli"
17		-	"bnema.dev/ermite/internal/adapters/httpserver"
18		-	"bnema.dev/ermite/internal/adapters/kvstore"
19		-	"bnema.dev/ermite/internal/app"
20	14		"bnema.dev/ermite/internal/config"
21	15		)
22	16

@@ -59,48 +53,14 @@ func serve(ctx context.Context) error {
59	53		if err := prepareDataDir(cfg.DataDir); err != nil {
60	54		return err
61	55		}
62		-	store, err := kvstore.Open(ctx, filepath.Join(cfg.DataDir, "kv"))
	56	+	f, err := newForge(ctx, cfg, options{})
63	57		if err != nil {
64	58		return err
65	59		}
66	60		// The server has stopped by the time this runs; close even when ctx ended.
67		-	defer func() {
68		-	if cerr := store.Close(context.WithoutCancel(ctx)); cerr != nil {
69		-	log.Error().Err(cerr).Msg("close metadata store")
70		-	}
71		-	}()
72		-	// The kv store holds the directory lock: only now is run/ ours to reset.
73		-	if err := resetRunDir(cfg.DataDir); err != nil {
74		-	return err
75		-	}
76		-
77		-	path := os.Getenv("PATH")
78		-	if path == "" {
79		-	path = "/usr/local/bin:/usr/bin:/bin"
80		-	}
81		-	git := gitcli.New(gitcli.Config{Bin: cfg.Git, DataDir: cfg.DataDir, Path: path})
82		-	gitVersion, err := git.CheckVersion(ctx)
83		-	if err != nil {
84		-	return fmt.Errorf("%s=%q: %w", config.EnvGit, cfg.Git, err)
85		-	}
86		-	log.Info().Str("git", gitVersion.String()).Msg("git checked")
87		-
88		-	repos := app.NewRepos(store, git, systemClock{})
89		-	report, err := repos.Reconcile(ctx)
90		-	if err != nil {
91		-	return fmt.Errorf("reconcile repositories: %w", err)
92		-	}
93		-	log.Info().
94		-	Int("registered", len(report.Registered)).
95		-	Int("missing_dirs", len(report.MissingDirs)).
96		-	Int("invalid_dirs", len(report.InvalidDirs)).
97		-	Msg("repositories reconciled")
	61	+	defer f.Close(context.WithoutCancel(ctx))
98	62
99		-	srv := httpserver.New(httpserver.Config{
100		-	Addr: cfg.Addr,
101		-	HealthChecks: []httpserver.Check{store.Check, httpserver.DirWritable(cfg.DataDir)},
102		-	})
103		-	if err := srv.Run(ctx); err != nil {
	63	+	if err := f.Run(ctx); err != nil {
104	64		return err
105	65		}
106	66		log.Info().Msg("stopped")


@@ -4,19 +4,13 @@ package main
4			4
5		import (	5		import (
6		"context"	6		"context"
7	-	"fmt"
8		"os"	7		"os"
9		"os/signal"	8		"os/signal"
10	-	"path/filepath"
11		"syscall"	9		"syscall"
12			10
13		"github.com/bnema/zerowrap"	11		"github.com/bnema/zerowrap"
14		"github.com/spf13/cobra"	12		"github.com/spf13/cobra"
15			13
16	-	"bnema.dev/ermite/internal/adapters/gitcli"
17	-	"bnema.dev/ermite/internal/adapters/httpserver"
18	-	"bnema.dev/ermite/internal/adapters/kvstore"
19	-	"bnema.dev/ermite/internal/app"
20		"bnema.dev/ermite/internal/config"	14		"bnema.dev/ermite/internal/config"
21		)	15		)
22			16

@@ -59,48 +53,14 @@ func serve(ctx context.Context) error {
59		if err := prepareDataDir(cfg.DataDir); err != nil {	53		if err := prepareDataDir(cfg.DataDir); err != nil {
60		return err	54		return err
61		}	55		}
62	-	store, err := kvstore.Open(ctx, filepath.Join(cfg.DataDir, "kv"))	56	+	f, err := newForge(ctx, cfg, options{})
63		if err != nil {	57		if err != nil {
64		return err	58		return err
65		}	59		}
66		// The server has stopped by the time this runs; close even when ctx ended.	60		// The server has stopped by the time this runs; close even when ctx ended.
67	-	defer func() {	61	+	defer f.Close(context.WithoutCancel(ctx))
68	-	if cerr := store.Close(context.WithoutCancel(ctx)); cerr != nil {
69	-	log.Error().Err(cerr).Msg("close metadata store")
70	-	}
71	-	}()
72	-	// The kv store holds the directory lock: only now is run/ ours to reset.
73	-	if err := resetRunDir(cfg.DataDir); err != nil {
74	-	return err
75	-	}
76	-
77	-	path := os.Getenv("PATH")
78	-	if path == "" {
79	-	path = "/usr/local/bin:/usr/bin:/bin"
80	-	}
81	-	git := gitcli.New(gitcli.Config{Bin: cfg.Git, DataDir: cfg.DataDir, Path: path})
82	-	gitVersion, err := git.CheckVersion(ctx)
83	-	if err != nil {
84	-	return fmt.Errorf("%s=%q: %w", config.EnvGit, cfg.Git, err)
85	-	}
86	-	log.Info().Str("git", gitVersion.String()).Msg("git checked")
87	-
88	-	repos := app.NewRepos(store, git, systemClock{})
89	-	report, err := repos.Reconcile(ctx)
90	-	if err != nil {
91	-	return fmt.Errorf("reconcile repositories: %w", err)
92	-	}
93	-	log.Info().
94	-	Int("registered", len(report.Registered)).
95	-	Int("missing_dirs", len(report.MissingDirs)).
96	-	Int("invalid_dirs", len(report.InvalidDirs)).
97	-	Msg("repositories reconciled")
98			62
99	-	srv := httpserver.New(httpserver.Config{	63	+	if err := f.Run(ctx); err != nil {
100	-	Addr: cfg.Addr,
101	-	HealthChecks: []httpserver.Check{store.Check, httpserver.DirWritable(cfg.DataDir)},
102	-	})
103	-	if err := srv.Run(ctx); err != nil {
104		return err	64		return err
105		}	65		}
106		log.Info().Msg("stopped")	66		log.Info().Msg("stopped")

A internal/adapters/eventbus/bus.go +90 -0

@@ -0,0 +1,90 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	// Package eventbus implements ports.EventBus: in-process fan-out to
	4	+	// subscribers, never blocking the publisher.
	5	+	package eventbus
	6	+
	7	+	import (
	8	+	"context"
	9	+	"sync"
	10	+
	11	+	"bnema.dev/ermite/internal/domain"
	12	+	"bnema.dev/ermite/internal/ports"
	13	+	)
	14	+
	15	+	// DefaultBuffer is the number of events a subscriber may lag behind.
	16	+	const DefaultBuffer = 64
	17	+
	18	+	var _ ports.EventBus = (*Bus)(nil)
	19	+
	20	+	// Bus is the in-process event bus.
	21	+	type Bus struct {
	22	+	buffer int
	23	+
	24	+	mu sync.Mutex
	25	+	subs map[*subscriber]struct{}
	26	+	}
	27	+
	28	+	type subscriber struct {
	29	+	ch chan domain.Event
	30	+	}
	31	+
	32	+	// New builds a bus whose subscribers buffer DefaultBuffer events.
	33	+	func New() *Bus { return NewWithBuffer(DefaultBuffer) }
	34	+
	35	+	// NewWithBuffer builds a bus whose subscribers buffer n events (at least 2,
	36	+	// so that a gap marker and the event that follows it fit).
	37	+	func NewWithBuffer(n int) *Bus {
	38	+	return &Bus{buffer: max(n, 2), subs: make(map[*subscriber]struct{})}
	39	+	}
	40	+
	41	+	// Publish delivers ev to every subscriber. A subscriber whose buffer is full
	42	+	// is cleared and gets a domain.EventGap followed by ev: it lost events and
	43	+	// must refresh its view, but it is never blocked on and never blocks others.
	44	+	func (b *Bus) Publish(_ context.Context, ev domain.Event) {
	45	+	b.mu.Lock()
	46	+	defer b.mu.Unlock()
	47	+	for s := range b.subs {
	48	+	select {
	49	+	case s.ch <- ev:
	50	+	continue
	51	+	default:
	52	+	}
	53	+	// Full. Only Publish sends and it holds b.mu, so after draining
	54	+	// there is room for the marker and the event.
	55	+	drain:
	56	+	for {
	57	+	select {
	58	+	case <-s.ch:
	59	+	default:
	60	+	break drain
	61	+	}
	62	+	}
	63	+	s.ch <- domain.EventGap{}
	64	+	s.ch <- ev
	65	+	}
	66	+	}
	67	+
	68	+	// Subscribe registers a subscriber until ctx ends, then closes its channel.
	69	+	func (b *Bus) Subscribe(ctx context.Context) <-chan domain.Event {
	70	+	s := &subscriber{ch: make(chan domain.Event, b.buffer)}
	71	+	b.mu.Lock()
	72	+	b.subs[s] = struct{}{}
	73	+	b.mu.Unlock()
	74	+
	75	+	go func() {
	76	+	<-ctx.Done()
	77	+	b.mu.Lock()
	78	+	delete(b.subs, s)
	79	+	close(s.ch)
	80	+	b.mu.Unlock()
	81	+	}()
	82	+	return s.ch
	83	+	}
	84	+
	85	+	// Subscribers returns the number of live subscribers, for tests.
	86	+	func (b *Bus) Subscribers() int {
	87	+	b.mu.Lock()
	88	+	defer b.mu.Unlock()
	89	+	return len(b.subs)
	90	+	}

@@ -0,0 +1,90 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	// Package eventbus implements ports.EventBus: in-process fan-out to
	4	+	// subscribers, never blocking the publisher.
	5	+	package eventbus
	6	+
	7	+	import (
	8	+	"context"
	9	+	"sync"
	10	+
	11	+	"bnema.dev/ermite/internal/domain"
	12	+	"bnema.dev/ermite/internal/ports"
	13	+	)
	14	+
	15	+	// DefaultBuffer is the number of events a subscriber may lag behind.
	16	+	const DefaultBuffer = 64
	17	+
	18	+	var _ ports.EventBus = (*Bus)(nil)
	19	+
	20	+	// Bus is the in-process event bus.
	21	+	type Bus struct {
	22	+	buffer int
	23	+
	24	+	mu sync.Mutex
	25	+	subs map[*subscriber]struct{}
	26	+	}
	27	+
	28	+	type subscriber struct {
	29	+	ch chan domain.Event
	30	+	}
	31	+
	32	+	// New builds a bus whose subscribers buffer DefaultBuffer events.
	33	+	func New() *Bus { return NewWithBuffer(DefaultBuffer) }
	34	+
	35	+	// NewWithBuffer builds a bus whose subscribers buffer n events (at least 2,
	36	+	// so that a gap marker and the event that follows it fit).
	37	+	func NewWithBuffer(n int) *Bus {
	38	+	return &Bus{buffer: max(n, 2), subs: make(map[*subscriber]struct{})}
	39	+	}
	40	+
	41	+	// Publish delivers ev to every subscriber. A subscriber whose buffer is full
	42	+	// is cleared and gets a domain.EventGap followed by ev: it lost events and
	43	+	// must refresh its view, but it is never blocked on and never blocks others.
	44	+	func (b *Bus) Publish(_ context.Context, ev domain.Event) {
	45	+	b.mu.Lock()
	46	+	defer b.mu.Unlock()
	47	+	for s := range b.subs {
	48	+	select {
	49	+	case s.ch <- ev:
	50	+	continue
	51	+	default:
	52	+	}
	53	+	// Full. Only Publish sends and it holds b.mu, so after draining
	54	+	// there is room for the marker and the event.
	55	+	drain:
	56	+	for {
	57	+	select {
	58	+	case <-s.ch:
	59	+	default:
	60	+	break drain
	61	+	}
	62	+	}
	63	+	s.ch <- domain.EventGap{}
	64	+	s.ch <- ev
	65	+	}
	66	+	}
	67	+
	68	+	// Subscribe registers a subscriber until ctx ends, then closes its channel.
	69	+	func (b *Bus) Subscribe(ctx context.Context) <-chan domain.Event {
	70	+	s := &subscriber{ch: make(chan domain.Event, b.buffer)}
	71	+	b.mu.Lock()
	72	+	b.subs[s] = struct{}{}
	73	+	b.mu.Unlock()
	74	+
	75	+	go func() {
	76	+	<-ctx.Done()
	77	+	b.mu.Lock()
	78	+	delete(b.subs, s)
	79	+	close(s.ch)
	80	+	b.mu.Unlock()
	81	+	}()
	82	+	return s.ch
	83	+	}
	84	+
	85	+	// Subscribers returns the number of live subscribers, for tests.
	86	+	func (b *Bus) Subscribers() int {
	87	+	b.mu.Lock()
	88	+	defer b.mu.Unlock()
	89	+	return len(b.subs)
	90	+	}

A internal/adapters/eventbus/bus_test.go +139 -0

@@ -0,0 +1,139 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package eventbus_test
	4	+
	5	+	import (
	6	+	"context"
	7	+	"sync"
	8	+	"testing"
	9	+	"time"
	10	+
	11	+	"github.com/stretchr/testify/assert"
	12	+	"github.com/stretchr/testify/require"
	13	+
	14	+	"bnema.dev/ermite/internal/adapters/eventbus"
	15	+	"bnema.dev/ermite/internal/domain"
	16	+	)
	17	+
	18	+	func ev(ref string) domain.Event {
	19	+	return domain.PushEvent{Updates: []domain.RefUpdate{{Ref: ref}}}
	20	+	}
	21	+
	22	+	func recv(t *testing.T, ch <-chan domain.Event) domain.Event {
	23	+	t.Helper()
	24	+	select {
	25	+	case e, ok := <-ch:
	26	+	require.True(t, ok, "channel closed")
	27	+	return e
	28	+	case <-time.After(2 * time.Second):
	29	+	t.Fatal("no event")
	30	+	return nil
	31	+	}
	32	+	}
	33	+
	34	+	func TestFanOut(t *testing.T) {
	35	+	ctx, cancel := context.WithCancel(context.Background())
	36	+	defer cancel()
	37	+	b := eventbus.New()
	38	+	a, c := b.Subscribe(ctx), b.Subscribe(ctx)
	39	+
	40	+	b.Publish(ctx, ev("1"))
	41	+	b.Publish(ctx, domain.RepoCreated{})
	42	+	assert.Equal(t, ev("1"), recv(t, a))
	43	+	assert.Equal(t, ev("1"), recv(t, c))
	44	+	assert.Equal(t, "repo.created", recv(t, a).EventKind())
	45	+	assert.Equal(t, "repo.created", recv(t, c).EventKind())
	46	+	}
	47	+
	48	+	func TestOnlyEventsFromSubscription(t *testing.T) {
	49	+	ctx, cancel := context.WithCancel(context.Background())
	50	+	defer cancel()
	51	+	b := eventbus.New()
	52	+	b.Publish(ctx, ev("before"))
	53	+	ch := b.Subscribe(ctx)
	54	+	b.Publish(ctx, ev("after"))
	55	+	assert.Equal(t, ev("after"), recv(t, ch))
	56	+	}
	57	+
	58	+	func TestSlowSubscriberGetsGapAndDoesNotBlock(t *testing.T) {
	59	+	ctx, cancel := context.WithCancel(context.Background())
	60	+	defer cancel()
	61	+	b := eventbus.NewWithBuffer(4)
	62	+	slow, fast := b.Subscribe(ctx), b.Subscribe(ctx)
	63	+
	64	+	done := make(chan struct{})
	65	+	go func() {
	66	+	defer close(done)
	67	+	for i := range 20 {
	68	+	b.Publish(ctx, ev(string(rune('a'+i))))
	69	+	// The fast subscriber keeps up.
	70	+	recv(t, fast)
	71	+	}
	72	+	}()
	73	+	select {
	74	+	case <-done:
	75	+	case <-time.After(5 * time.Second):
	76	+	t.Fatal("Publish blocked on a slow subscriber")
	77	+	}
	78	+
	79	+	// The slow one lost events: it sees a gap marker, then only recent events.
	80	+	var got []domain.Event
	81	+	for len(slow) > 0 {
	82	+	got = append(got, <-slow)
	83	+	}
	84	+	require.NotEmpty(t, got)
	85	+	var gapAt = -1
	86	+	for i, e := range got {
	87	+	if _, ok := e.(domain.EventGap); ok {
	88	+	gapAt = i
	89	+	}
	90	+	}
	91	+	assert.GreaterOrEqual(t, gapAt, 0, "a gap marker is delivered")
	92	+	assert.Equal(t, ev("t"), got[len(got)-1], "the latest event is delivered")
	93	+	assert.LessOrEqual(t, len(got), 4)
	94	+	}
	95	+
	96	+	func TestUnsubscribeOnContext(t *testing.T) {
	97	+	b := eventbus.New()
	98	+	ctx, cancel := context.WithCancel(context.Background())
	99	+	ch := b.Subscribe(ctx)
	100	+	require.Equal(t, 1, b.Subscribers())
	101	+	cancel()
	102	+	select {
	103	+	case _, ok := <-ch:
	104	+	assert.False(t, ok)
	105	+	case <-time.After(2 * time.Second):
	106	+	t.Fatal("channel not closed")
	107	+	}
	108	+	assert.Equal(t, 0, b.Subscribers())
	109	+	b.Publish(context.Background(), ev("x")) // no panic
	110	+	}
	111	+
	112	+	func TestConcurrentPublishSubscribe(t *testing.T) {
	113	+	b := eventbus.NewWithBuffer(8)
	114	+	ctx, cancel := context.WithCancel(context.Background())
	115	+	var wg sync.WaitGroup
	116	+	for range 4 {
	117	+	wg.Add(2)
	118	+	go func() {
	119	+	defer wg.Done()
	120	+	for range 200 {
	121	+	b.Publish(ctx, ev("x"))
	122	+	}
	123	+	}()
	124	+	go func() {
	125	+	defer wg.Done()
	126	+	sctx, scancel := context.WithCancel(ctx)
	127	+	ch := b.Subscribe(sctx)
	128	+	for range 5 {
	129	+	select {
	130	+	case <-ch:
	131	+	case <-time.After(10 * time.Millisecond):
	132	+	}
	133	+	}
	134	+	scancel()
	135	+	}()
	136	+	}
	137	+	wg.Wait()
	138	+	cancel()
	139	+	}

@@ -0,0 +1,139 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package eventbus_test
	4	+
	5	+	import (
	6	+	"context"
	7	+	"sync"
	8	+	"testing"
	9	+	"time"
	10	+
	11	+	"github.com/stretchr/testify/assert"
	12	+	"github.com/stretchr/testify/require"
	13	+
	14	+	"bnema.dev/ermite/internal/adapters/eventbus"
	15	+	"bnema.dev/ermite/internal/domain"
	16	+	)
	17	+
	18	+	func ev(ref string) domain.Event {
	19	+	return domain.PushEvent{Updates: []domain.RefUpdate{{Ref: ref}}}
	20	+	}
	21	+
	22	+	func recv(t *testing.T, ch <-chan domain.Event) domain.Event {
	23	+	t.Helper()
	24	+	select {
	25	+	case e, ok := <-ch:
	26	+	require.True(t, ok, "channel closed")
	27	+	return e
	28	+	case <-time.After(2 * time.Second):
	29	+	t.Fatal("no event")
	30	+	return nil
	31	+	}
	32	+	}
	33	+
	34	+	func TestFanOut(t *testing.T) {
	35	+	ctx, cancel := context.WithCancel(context.Background())
	36	+	defer cancel()
	37	+	b := eventbus.New()
	38	+	a, c := b.Subscribe(ctx), b.Subscribe(ctx)
	39	+
	40	+	b.Publish(ctx, ev("1"))
	41	+	b.Publish(ctx, domain.RepoCreated{})
	42	+	assert.Equal(t, ev("1"), recv(t, a))
	43	+	assert.Equal(t, ev("1"), recv(t, c))
	44	+	assert.Equal(t, "repo.created", recv(t, a).EventKind())
	45	+	assert.Equal(t, "repo.created", recv(t, c).EventKind())
	46	+	}
	47	+
	48	+	func TestOnlyEventsFromSubscription(t *testing.T) {
	49	+	ctx, cancel := context.WithCancel(context.Background())
	50	+	defer cancel()
	51	+	b := eventbus.New()
	52	+	b.Publish(ctx, ev("before"))
	53	+	ch := b.Subscribe(ctx)
	54	+	b.Publish(ctx, ev("after"))
	55	+	assert.Equal(t, ev("after"), recv(t, ch))
	56	+	}
	57	+
	58	+	func TestSlowSubscriberGetsGapAndDoesNotBlock(t *testing.T) {
	59	+	ctx, cancel := context.WithCancel(context.Background())
	60	+	defer cancel()
	61	+	b := eventbus.NewWithBuffer(4)
	62	+	slow, fast := b.Subscribe(ctx), b.Subscribe(ctx)
	63	+
	64	+	done := make(chan struct{})
	65	+	go func() {
	66	+	defer close(done)
	67	+	for i := range 20 {
	68	+	b.Publish(ctx, ev(string(rune('a'+i))))
	69	+	// The fast subscriber keeps up.
	70	+	recv(t, fast)
	71	+	}
	72	+	}()
	73	+	select {
	74	+	case <-done:
	75	+	case <-time.After(5 * time.Second):
	76	+	t.Fatal("Publish blocked on a slow subscriber")
	77	+	}
	78	+
	79	+	// The slow one lost events: it sees a gap marker, then only recent events.
	80	+	var got []domain.Event
	81	+	for len(slow) > 0 {
	82	+	got = append(got, <-slow)
	83	+	}
	84	+	require.NotEmpty(t, got)
	85	+	var gapAt = -1
	86	+	for i, e := range got {
	87	+	if _, ok := e.(domain.EventGap); ok {
	88	+	gapAt = i
	89	+	}
	90	+	}
	91	+	assert.GreaterOrEqual(t, gapAt, 0, "a gap marker is delivered")
	92	+	assert.Equal(t, ev("t"), got[len(got)-1], "the latest event is delivered")
	93	+	assert.LessOrEqual(t, len(got), 4)
	94	+	}
	95	+
	96	+	func TestUnsubscribeOnContext(t *testing.T) {
	97	+	b := eventbus.New()
	98	+	ctx, cancel := context.WithCancel(context.Background())
	99	+	ch := b.Subscribe(ctx)
	100	+	require.Equal(t, 1, b.Subscribers())
	101	+	cancel()
	102	+	select {
	103	+	case _, ok := <-ch:
	104	+	assert.False(t, ok)
	105	+	case <-time.After(2 * time.Second):
	106	+	t.Fatal("channel not closed")
	107	+	}
	108	+	assert.Equal(t, 0, b.Subscribers())
	109	+	b.Publish(context.Background(), ev("x")) // no panic
	110	+	}
	111	+
	112	+	func TestConcurrentPublishSubscribe(t *testing.T) {
	113	+	b := eventbus.NewWithBuffer(8)
	114	+	ctx, cancel := context.WithCancel(context.Background())
	115	+	var wg sync.WaitGroup
	116	+	for range 4 {
	117	+	wg.Add(2)
	118	+	go func() {
	119	+	defer wg.Done()
	120	+	for range 200 {
	121	+	b.Publish(ctx, ev("x"))
	122	+	}
	123	+	}()
	124	+	go func() {
	125	+	defer wg.Done()
	126	+	sctx, scancel := context.WithCancel(ctx)
	127	+	ch := b.Subscribe(sctx)
	128	+	for range 5 {
	129	+	select {
	130	+	case <-ch:
	131	+	case <-time.After(10 * time.Millisecond):
	132	+	}
	133	+	}
	134	+	scancel()
	135	+	}()
	136	+	}
	137	+	wg.Wait()
	138	+	cancel()
	139	+	}

A internal/app/maintenance.go +123 -0

@@ -0,0 +1,123 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package app
	4	+
	5	+	import (
	6	+	"context"
	7	+	"sync"
	8	+	"time"
	9	+
	10	+	"github.com/bnema/zerowrap"
	11	+
	12	+	"bnema.dev/ermite/internal/domain"
	13	+	"bnema.dev/ermite/internal/ports"
	14	+	)
	15	+
	16	+	// MaintenanceDelay is how long maintenance waits after the first push of a
	17	+	// burst, so that pushes in quick succession share one run.
	18	+	const MaintenanceDelay = 5 * time.Second
	19	+
	20	+	// Maintainer runs repository maintenance after pushes, debounced per
	21	+	// repository: the first Schedule starts a timer, further ones before it fires
	22	+	// join the same run, and a repository is maintained by one run at a time. A
	23	+	// Schedule during a run queues exactly one more run.
	24	+	//
	25	+	// Its goroutines use the logger of the context given to NewMaintainer and
	26	+	// live until Close.
	27	+	type Maintainer struct {
	28	+	git ports.GitRepos
	29	+	delay time.Duration
	30	+	ctx context.Context
	31	+	stop context.CancelFunc
	32	+
	33	+	mu sync.Mutex
	34	+	closed bool
	35	+	entries map[domain.RepoName]*maintEntry
	36	+	wg sync.WaitGroup
	37	+	}
	38	+
	39	+	type maintEntry struct {
	40	+	timer *time.Timer // pending run, nil while running or idle
	41	+	running bool
	42	+	again bool // scheduled during a run
	43	+	}
	44	+
	45	+	// NewMaintainer builds a maintainer. ctx carries the logger; its cancellation
	46	+	// is not followed, call Close. A zero delay means MaintenanceDelay.
	47	+	func NewMaintainer(ctx context.Context, git ports.GitRepos, delay time.Duration) *Maintainer {
	48	+	if delay <= 0 {
	49	+	delay = MaintenanceDelay
	50	+	}
	51	+	run, stop := context.WithCancel(context.WithoutCancel(ctx))
	52	+	return &Maintainer{git: git, delay: delay, ctx: run, stop: stop, entries: make(map[domain.RepoName]*maintEntry)}
	53	+	}
	54	+
	55	+	// Schedule queues maintenance of the repository.
	56	+	func (m *Maintainer) Schedule(name domain.RepoName) {
	57	+	m.mu.Lock()
	58	+	defer m.mu.Unlock()
	59	+	if m.closed {
	60	+	return
	61	+	}
	62	+	e := m.entries[name]
	63	+	switch {
	64	+	case e == nil:
	65	+	e = &maintEntry{}
	66	+	m.entries[name] = e
	67	+	m.arm(name, e)
	68	+	case e.running:
	69	+	e.again = true
	70	+	}
	71	+	// A timer is already pending: this push joins it.
	72	+	}
	73	+
	74	+	// arm starts the timer of e. The caller holds mu.
	75	+	func (m Maintainer) arm(name domain.RepoName, e maintEntry) {
	76	+	m.wg.Add(1) // released when the timer fires or is stopped by Close
	77	+	e.timer = time.AfterFunc(m.delay, func() { m.run(name, e) })
	78	+	}
	79	+
	80	+	func (m Maintainer) run(name domain.RepoName, e maintEntry) {
	81	+	defer m.wg.Done()
	82	+	m.mu.Lock()
	83	+	if m.closed {
	84	+	delete(m.entries, name)
	85	+	m.mu.Unlock()
	86	+	return
	87	+	}
	88	+	e.timer, e.running = nil, true
	89	+	m.mu.Unlock()
	90	+
	91	+	log := zerowrap.FromCtx(m.ctx)
	92	+	start := time.Now()
	93	+	if err := m.git.Maintain(m.ctx, name); err != nil {
	94	+	log.Warn().Err(err).Str("repo", name.String()).Msg("repository maintenance failed")
	95	+	} else {
	96	+	log.Debug().Str("repo", name.String()).Dur("duration", time.Since(start)).Msg("repository maintained")
	97	+	}
	98	+
	99	+	m.mu.Lock()
	100	+	defer m.mu.Unlock()
	101	+	e.running = false
	102	+	if e.again && !m.closed {
	103	+	e.again = false
	104	+	m.arm(name, e)
	105	+	return
	106	+	}
	107	+	delete(m.entries, name)
	108	+	}
	109	+
	110	+	// Close stops pending runs, cancels the one in progress and waits for it.
	111	+	func (m *Maintainer) Close(_ context.Context) {
	112	+	m.mu.Lock()
	113	+	m.closed = true
	114	+	for _, e := range m.entries {
	115	+	if e.timer != nil && e.timer.Stop() {
	116	+	m.wg.Done() // the timer will never fire
	117	+	e.timer = nil
	118	+	}
	119	+	}
	120	+	m.mu.Unlock()
	121	+	m.stop()
	122	+	m.wg.Wait()
	123	+	}

@@ -0,0 +1,123 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package app
	4	+
	5	+	import (
	6	+	"context"
	7	+	"sync"
	8	+	"time"
	9	+
	10	+	"github.com/bnema/zerowrap"
	11	+
	12	+	"bnema.dev/ermite/internal/domain"
	13	+	"bnema.dev/ermite/internal/ports"
	14	+	)
	15	+
	16	+	// MaintenanceDelay is how long maintenance waits after the first push of a
	17	+	// burst, so that pushes in quick succession share one run.
	18	+	const MaintenanceDelay = 5 * time.Second
	19	+
	20	+	// Maintainer runs repository maintenance after pushes, debounced per
	21	+	// repository: the first Schedule starts a timer, further ones before it fires
	22	+	// join the same run, and a repository is maintained by one run at a time. A
	23	+	// Schedule during a run queues exactly one more run.
	24	+	//
	25	+	// Its goroutines use the logger of the context given to NewMaintainer and
	26	+	// live until Close.
	27	+	type Maintainer struct {
	28	+	git ports.GitRepos
	29	+	delay time.Duration
	30	+	ctx context.Context
	31	+	stop context.CancelFunc
	32	+
	33	+	mu sync.Mutex
	34	+	closed bool
	35	+	entries map[domain.RepoName]*maintEntry
	36	+	wg sync.WaitGroup
	37	+	}
	38	+
	39	+	type maintEntry struct {
	40	+	timer *time.Timer // pending run, nil while running or idle
	41	+	running bool
	42	+	again bool // scheduled during a run
	43	+	}
	44	+
	45	+	// NewMaintainer builds a maintainer. ctx carries the logger; its cancellation
	46	+	// is not followed, call Close. A zero delay means MaintenanceDelay.
	47	+	func NewMaintainer(ctx context.Context, git ports.GitRepos, delay time.Duration) *Maintainer {
	48	+	if delay <= 0 {
	49	+	delay = MaintenanceDelay
	50	+	}
	51	+	run, stop := context.WithCancel(context.WithoutCancel(ctx))
	52	+	return &Maintainer{git: git, delay: delay, ctx: run, stop: stop, entries: make(map[domain.RepoName]*maintEntry)}
	53	+	}
	54	+
	55	+	// Schedule queues maintenance of the repository.
	56	+	func (m *Maintainer) Schedule(name domain.RepoName) {
	57	+	m.mu.Lock()
	58	+	defer m.mu.Unlock()
	59	+	if m.closed {
	60	+	return
	61	+	}
	62	+	e := m.entries[name]
	63	+	switch {
	64	+	case e == nil:
	65	+	e = &maintEntry{}
	66	+	m.entries[name] = e
	67	+	m.arm(name, e)
	68	+	case e.running:
	69	+	e.again = true
	70	+	}
	71	+	// A timer is already pending: this push joins it.
	72	+	}
	73	+
	74	+	// arm starts the timer of e. The caller holds mu.
	75	+	func (m Maintainer) arm(name domain.RepoName, e maintEntry) {
	76	+	m.wg.Add(1) // released when the timer fires or is stopped by Close
	77	+	e.timer = time.AfterFunc(m.delay, func() { m.run(name, e) })
	78	+	}
	79	+
	80	+	func (m Maintainer) run(name domain.RepoName, e maintEntry) {
	81	+	defer m.wg.Done()
	82	+	m.mu.Lock()
	83	+	if m.closed {
	84	+	delete(m.entries, name)
	85	+	m.mu.Unlock()
	86	+	return
	87	+	}
	88	+	e.timer, e.running = nil, true
	89	+	m.mu.Unlock()
	90	+
	91	+	log := zerowrap.FromCtx(m.ctx)
	92	+	start := time.Now()
	93	+	if err := m.git.Maintain(m.ctx, name); err != nil {
	94	+	log.Warn().Err(err).Str("repo", name.String()).Msg("repository maintenance failed")
	95	+	} else {
	96	+	log.Debug().Str("repo", name.String()).Dur("duration", time.Since(start)).Msg("repository maintained")
	97	+	}
	98	+
	99	+	m.mu.Lock()
	100	+	defer m.mu.Unlock()
	101	+	e.running = false
	102	+	if e.again && !m.closed {
	103	+	e.again = false
	104	+	m.arm(name, e)
	105	+	return
	106	+	}
	107	+	delete(m.entries, name)
	108	+	}
	109	+
	110	+	// Close stops pending runs, cancels the one in progress and waits for it.
	111	+	func (m *Maintainer) Close(_ context.Context) {
	112	+	m.mu.Lock()
	113	+	m.closed = true
	114	+	for _, e := range m.entries {
	115	+	if e.timer != nil && e.timer.Stop() {
	116	+	m.wg.Done() // the timer will never fire
	117	+	e.timer = nil
	118	+	}
	119	+	}
	120	+	m.mu.Unlock()
	121	+	m.stop()
	122	+	m.wg.Wait()
	123	+	}

A internal/app/maintenance_test.go +174 -0

@@ -0,0 +1,174 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package app_test
	4	+
	5	+	import (
	6	+	"context"
	7	+	"errors"
	8	+	"sync"
	9	+	"sync/atomic"
	10	+	"testing"
	11	+	"time"
	12	+
	13	+	"github.com/stretchr/testify/assert"
	14	+	"github.com/stretchr/testify/require"
	15	+
	16	+	"bnema.dev/ermite/internal/app"
	17	+	"bnema.dev/ermite/internal/domain"
	18	+	"bnema.dev/ermite/internal/ports/mocks"
	19	+	)
	20	+
	21	+	func waitFor(t *testing.T, cond func() bool, what string) {
	22	+	t.Helper()
	23	+	deadline := time.Now().Add(5 * time.Second)
	24	+	for !cond() {
	25	+	if time.Now().After(deadline) {
	26	+	t.Fatalf("timeout waiting for %s", what)
	27	+	}
	28	+	time.Sleep(2 * time.Millisecond)
	29	+	}
	30	+	}
	31	+
	32	+	func TestMaintainerDebounces(t *testing.T) {
	33	+	ctx := context.Background()
	34	+	n := name(t, "tools")
	35	+	git := mocks.NewGitRepos(t)
	36	+	var runs atomic.Int32
	37	+	git.EXPECT().Maintain(mockAny, n).RunAndReturn(func(context.Context, domain.RepoName) error {
	38	+	runs.Add(1)
	39	+	return nil
	40	+	})
	41	+	m := app.NewMaintainer(ctx, git, 50*time.Millisecond)
	42	+	defer m.Close(ctx)
	43	+
	44	+	for range 10 {
	45	+	m.Schedule(n)
	46	+	}
	47	+	time.Sleep(10 * time.Millisecond)
	48	+	assert.Equal(t, int32(0), runs.Load(), "waits for the delay")
	49	+	waitFor(t, func() bool { return runs.Load() == 1 }, "one run")
	50	+	time.Sleep(150 * time.Millisecond)
	51	+	assert.Equal(t, int32(1), runs.Load(), "a burst makes one run")
	52	+
	53	+	m.Schedule(n)
	54	+	waitFor(t, func() bool { return runs.Load() == 2 }, "a second burst, a second run")
	55	+	}
	56	+
	57	+	func TestMaintainerOneRunAtATimePerRepo(t *testing.T) {
	58	+	ctx := context.Background()
	59	+	n, other := name(t, "tools"), name(t, "other")
	60	+	git := mocks.NewGitRepos(t)
	61	+	var concurrent, maxConcurrent, runs atomic.Int32
	62	+	release := make(chan struct{})
	63	+	var otherRan atomic.Bool
	64	+	git.EXPECT().Maintain(mockAny, n).RunAndReturn(func(context.Context, domain.RepoName) error {
	65	+	c := concurrent.Add(1)
	66	+	for {
	67	+	m := maxConcurrent.Load()
	68	+	if c <= m \|\| maxConcurrent.CompareAndSwap(m, c) {
	69	+	break
	70	+	}
	71	+	}
	72	+	runs.Add(1)
	73	+	<-release
	74	+	concurrent.Add(-1)
	75	+	return nil
	76	+	})
	77	+	git.EXPECT().Maintain(mockAny, other).RunAndReturn(func(context.Context, domain.RepoName) error {
	78	+	otherRan.Store(true)
	79	+	return nil
	80	+	})
	81	+	m := app.NewMaintainer(ctx, git, 10*time.Millisecond)
	82	+	defer m.Close(ctx)
	83	+
	84	+	m.Schedule(n)
	85	+	waitFor(t, func() bool { return runs.Load() == 1 }, "first run to start")
	86	+	m.Schedule(n) // during the run: queues exactly one more
	87	+	m.Schedule(n)
	88	+	m.Schedule(other)
	89	+	waitFor(t, otherRan.Load, "another repository is not blocked")
	90	+	time.Sleep(60 * time.Millisecond)
	91	+	assert.Equal(t, int32(1), runs.Load(), "the second run waits for the first")
	92	+
	93	+	release <- struct{}{} // finish first run
	94	+	waitFor(t, func() bool { return runs.Load() == 2 }, "queued run")
	95	+	release <- struct{}{}
	96	+	time.Sleep(60 * time.Millisecond)
	97	+	assert.Equal(t, int32(2), runs.Load(), "exactly one queued run")
	98	+	assert.Equal(t, int32(1), maxConcurrent.Load())
	99	+	}
	100	+
	101	+	func TestMaintainerFailureIsLoggedNotFatal(t *testing.T) {
	102	+	ctx := context.Background()
	103	+	n := name(t, "tools")
	104	+	git := mocks.NewGitRepos(t)
	105	+	var runs atomic.Int32
	106	+	git.EXPECT().Maintain(mockAny, n).RunAndReturn(func(context.Context, domain.RepoName) error {
	107	+	runs.Add(1)
	108	+	return errors.New("boom")
	109	+	})
	110	+	m := app.NewMaintainer(ctx, git, 5*time.Millisecond)
	111	+	defer m.Close(ctx)
	112	+	m.Schedule(n)
	113	+	waitFor(t, func() bool { return runs.Load() == 1 }, "run")
	114	+	m.Schedule(n)
	115	+	waitFor(t, func() bool { return runs.Load() == 2 }, "later runs still happen")
	116	+	}
	117	+
	118	+	func TestMaintainerClose(t *testing.T) {
	119	+	ctx := context.Background()
	120	+	n := name(t, "tools")
	121	+
	122	+	t.Run("pending runs are dropped", func(t *testing.T) {
	123	+	git := mocks.NewGitRepos(t) // no Maintain expected
	124	+	m := app.NewMaintainer(ctx, git, time.Hour)
	125	+	m.Schedule(n)
	126	+	m.Close(ctx)
	127	+	m.Schedule(n) // after Close: ignored
	128	+	m.Close(ctx) // twice is fine
	129	+	})
	130	+
	131	+	t.Run("a run in progress is cancelled and awaited", func(t *testing.T) {
	132	+	git := mocks.NewGitRepos(t)
	133	+	started := make(chan struct{})
	134	+	var cancelled atomic.Bool
	135	+	git.EXPECT().Maintain(mockAny, n).RunAndReturn(func(ctx context.Context, _ domain.RepoName) error {
	136	+	close(started)
	137	+	<-ctx.Done()
	138	+	cancelled.Store(true)
	139	+	return ctx.Err()
	140	+	})
	141	+	m := app.NewMaintainer(ctx, git, time.Millisecond)
	142	+	m.Schedule(n)
	143	+	<-started
	144	+	done := make(chan struct{})
	145	+	go func() { m.Close(ctx); close(done) }()
	146	+	select {
	147	+	case <-done:
	148	+	case <-time.After(5 * time.Second):
	149	+	t.Fatal("Close did not return")
	150	+	}
	151	+	assert.True(t, cancelled.Load())
	152	+	})
	153	+	}
	154	+
	155	+	func TestMaintainerConcurrentSchedule(t *testing.T) {
	156	+	ctx := context.Background()
	157	+	git := mocks.NewGitRepos(t)
	158	+	git.EXPECT().Maintain(mockAny, mockAny).Return(nil).Maybe()
	159	+	m := app.NewMaintainer(ctx, git, time.Millisecond)
	160	+	var wg sync.WaitGroup
	161	+	for i := range 8 {
	162	+	wg.Add(1)
	163	+	go func() {
	164	+	defer wg.Done()
	165	+	for range 50 {
	166	+	m.Schedule(name(t, string(rune('a'+i%3))))
	167	+	time.Sleep(100 * time.Microsecond)
	168	+	}
	169	+	}()
	170	+	}
	171	+	wg.Wait()
	172	+	m.Close(ctx)
	173	+	require.True(t, true)
	174	+	}

@@ -0,0 +1,174 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package app_test
	4	+
	5	+	import (
	6	+	"context"
	7	+	"errors"
	8	+	"sync"
	9	+	"sync/atomic"
	10	+	"testing"
	11	+	"time"
	12	+
	13	+	"github.com/stretchr/testify/assert"
	14	+	"github.com/stretchr/testify/require"
	15	+
	16	+	"bnema.dev/ermite/internal/app"
	17	+	"bnema.dev/ermite/internal/domain"
	18	+	"bnema.dev/ermite/internal/ports/mocks"
	19	+	)
	20	+
	21	+	func waitFor(t *testing.T, cond func() bool, what string) {
	22	+	t.Helper()
	23	+	deadline := time.Now().Add(5 * time.Second)
	24	+	for !cond() {
	25	+	if time.Now().After(deadline) {
	26	+	t.Fatalf("timeout waiting for %s", what)
	27	+	}
	28	+	time.Sleep(2 * time.Millisecond)
	29	+	}
	30	+	}
	31	+
	32	+	func TestMaintainerDebounces(t *testing.T) {
	33	+	ctx := context.Background()
	34	+	n := name(t, "tools")
	35	+	git := mocks.NewGitRepos(t)
	36	+	var runs atomic.Int32
	37	+	git.EXPECT().Maintain(mockAny, n).RunAndReturn(func(context.Context, domain.RepoName) error {
	38	+	runs.Add(1)
	39	+	return nil
	40	+	})
	41	+	m := app.NewMaintainer(ctx, git, 50*time.Millisecond)
	42	+	defer m.Close(ctx)
	43	+
	44	+	for range 10 {
	45	+	m.Schedule(n)
	46	+	}
	47	+	time.Sleep(10 * time.Millisecond)
	48	+	assert.Equal(t, int32(0), runs.Load(), "waits for the delay")
	49	+	waitFor(t, func() bool { return runs.Load() == 1 }, "one run")
	50	+	time.Sleep(150 * time.Millisecond)
	51	+	assert.Equal(t, int32(1), runs.Load(), "a burst makes one run")
	52	+
	53	+	m.Schedule(n)
	54	+	waitFor(t, func() bool { return runs.Load() == 2 }, "a second burst, a second run")
	55	+	}
	56	+
	57	+	func TestMaintainerOneRunAtATimePerRepo(t *testing.T) {
	58	+	ctx := context.Background()
	59	+	n, other := name(t, "tools"), name(t, "other")
	60	+	git := mocks.NewGitRepos(t)
	61	+	var concurrent, maxConcurrent, runs atomic.Int32
	62	+	release := make(chan struct{})
	63	+	var otherRan atomic.Bool
	64	+	git.EXPECT().Maintain(mockAny, n).RunAndReturn(func(context.Context, domain.RepoName) error {
	65	+	c := concurrent.Add(1)
	66	+	for {
	67	+	m := maxConcurrent.Load()
	68	+	if c <= m \|\| maxConcurrent.CompareAndSwap(m, c) {
	69	+	break
	70	+	}
	71	+	}
	72	+	runs.Add(1)
	73	+	<-release
	74	+	concurrent.Add(-1)
	75	+	return nil
	76	+	})
	77	+	git.EXPECT().Maintain(mockAny, other).RunAndReturn(func(context.Context, domain.RepoName) error {
	78	+	otherRan.Store(true)
	79	+	return nil
	80	+	})
	81	+	m := app.NewMaintainer(ctx, git, 10*time.Millisecond)
	82	+	defer m.Close(ctx)
	83	+
	84	+	m.Schedule(n)
	85	+	waitFor(t, func() bool { return runs.Load() == 1 }, "first run to start")
	86	+	m.Schedule(n) // during the run: queues exactly one more
	87	+	m.Schedule(n)
	88	+	m.Schedule(other)
	89	+	waitFor(t, otherRan.Load, "another repository is not blocked")
	90	+	time.Sleep(60 * time.Millisecond)
	91	+	assert.Equal(t, int32(1), runs.Load(), "the second run waits for the first")
	92	+
	93	+	release <- struct{}{} // finish first run
	94	+	waitFor(t, func() bool { return runs.Load() == 2 }, "queued run")
	95	+	release <- struct{}{}
	96	+	time.Sleep(60 * time.Millisecond)
	97	+	assert.Equal(t, int32(2), runs.Load(), "exactly one queued run")
	98	+	assert.Equal(t, int32(1), maxConcurrent.Load())
	99	+	}
	100	+
	101	+	func TestMaintainerFailureIsLoggedNotFatal(t *testing.T) {
	102	+	ctx := context.Background()
	103	+	n := name(t, "tools")
	104	+	git := mocks.NewGitRepos(t)
	105	+	var runs atomic.Int32
	106	+	git.EXPECT().Maintain(mockAny, n).RunAndReturn(func(context.Context, domain.RepoName) error {
	107	+	runs.Add(1)
	108	+	return errors.New("boom")
	109	+	})
	110	+	m := app.NewMaintainer(ctx, git, 5*time.Millisecond)
	111	+	defer m.Close(ctx)
	112	+	m.Schedule(n)
	113	+	waitFor(t, func() bool { return runs.Load() == 1 }, "run")
	114	+	m.Schedule(n)
	115	+	waitFor(t, func() bool { return runs.Load() == 2 }, "later runs still happen")
	116	+	}
	117	+
	118	+	func TestMaintainerClose(t *testing.T) {
	119	+	ctx := context.Background()
	120	+	n := name(t, "tools")
	121	+
	122	+	t.Run("pending runs are dropped", func(t *testing.T) {
	123	+	git := mocks.NewGitRepos(t) // no Maintain expected
	124	+	m := app.NewMaintainer(ctx, git, time.Hour)
	125	+	m.Schedule(n)
	126	+	m.Close(ctx)
	127	+	m.Schedule(n) // after Close: ignored
	128	+	m.Close(ctx) // twice is fine
	129	+	})
	130	+
	131	+	t.Run("a run in progress is cancelled and awaited", func(t *testing.T) {
	132	+	git := mocks.NewGitRepos(t)
	133	+	started := make(chan struct{})
	134	+	var cancelled atomic.Bool
	135	+	git.EXPECT().Maintain(mockAny, n).RunAndReturn(func(ctx context.Context, _ domain.RepoName) error {
	136	+	close(started)
	137	+	<-ctx.Done()
	138	+	cancelled.Store(true)
	139	+	return ctx.Err()
	140	+	})
	141	+	m := app.NewMaintainer(ctx, git, time.Millisecond)
	142	+	m.Schedule(n)
	143	+	<-started
	144	+	done := make(chan struct{})
	145	+	go func() { m.Close(ctx); close(done) }()
	146	+	select {
	147	+	case <-done:
	148	+	case <-time.After(5 * time.Second):
	149	+	t.Fatal("Close did not return")
	150	+	}
	151	+	assert.True(t, cancelled.Load())
	152	+	})
	153	+	}
	154	+
	155	+	func TestMaintainerConcurrentSchedule(t *testing.T) {
	156	+	ctx := context.Background()
	157	+	git := mocks.NewGitRepos(t)
	158	+	git.EXPECT().Maintain(mockAny, mockAny).Return(nil).Maybe()
	159	+	m := app.NewMaintainer(ctx, git, time.Millisecond)
	160	+	var wg sync.WaitGroup
	161	+	for i := range 8 {
	162	+	wg.Add(1)
	163	+	go func() {
	164	+	defer wg.Done()
	165	+	for range 50 {
	166	+	m.Schedule(name(t, string(rune('a'+i%3))))
	167	+	time.Sleep(100 * time.Microsecond)
	168	+	}
	169	+	}()
	170	+	}
	171	+	wg.Wait()
	172	+	m.Close(ctx)
	173	+	require.True(t, true)
	174	+	}