798e82d fix(mirror): return manual syncs within 8 s and run sync --all concurrently · ermite

M docs/mirror.md +1 -0


@@ -44,6 +44,7 @@ URLs are `git@host:path` or `ssh://[user@]host[:port]/path`, with no password. T
44	44
45	45		- On the cron schedule. A push to ermite only marks the repository as changed; the next due cron tick syncs it. A tick with nothing new does nothing.
46	46		- `ermite mirror sync <repo>` or `ermite mirror sync --all` (owner only) syncs now; a repository needs admin rights. `--all` skips repositories without a mirror, continues past failures, prints one row per repository and exits with status 1 if any failed.
	47	+	- A manual sync waits at most about 8 s (the edge proxy cuts responses at 10 s, the CLI at 30 s, while an SSH push to an unreachable host can take 30 s or more). `--all` starts every repository at once, so the wait is one push, not one per repository. A push still running then is shown as `QUEUED true`; the CLI says so on stderr, the sync continues on the server, and `ermite mirror status` shows its outcome.
47	48		- `--force` skips the no-op check and the 60 s throttle and lifts a pause. It never force-pushes.
48	49		- Overlapping syncs of a repository coalesce; only one push runs at a time per repository.
49	50


@@ -44,6 +44,7 @@ URLs are `git@host:path` or `ssh://[user@]host[:port]/path`, with no password. T
44		44
45	- On the cron schedule. A push to ermite only marks the repository as changed; the next due cron tick syncs it. A tick with nothing new does nothing.	45		- On the cron schedule. A push to ermite only marks the repository as changed; the next due cron tick syncs it. A tick with nothing new does nothing.
46	- `ermite mirror sync <repo>` or `ermite mirror sync --all` (owner only) syncs now; a repository needs admin rights. `--all` skips repositories without a mirror, continues past failures, prints one row per repository and exits with status 1 if any failed.	46		- `ermite mirror sync <repo>` or `ermite mirror sync --all` (owner only) syncs now; a repository needs admin rights. `--all` skips repositories without a mirror, continues past failures, prints one row per repository and exits with status 1 if any failed.
		47	+	- A manual sync waits at most about 8 s (the edge proxy cuts responses at 10 s, the CLI at 30 s, while an SSH push to an unreachable host can take 30 s or more). `--all` starts every repository at once, so the wait is one push, not one per repository. A push still running then is shown as `QUEUED true`; the CLI says so on stderr, the sync continues on the server, and `ermite mirror status` shows its outcome.
47	- `--force` skips the no-op check and the 60 s throttle and lifts a pause. It never force-pushes.	48		- `--force` skips the no-op check and the 60 s throttle and lifts a pause. It never force-pushes.
48	- Overlapping syncs of a repository coalesce; only one push runs at a time per repository.	49		- Overlapping syncs of a repository coalesce; only one push runs at a time per repository.
49		50

M internal/app/cli_mirror.go +8 -2


@@ -277,7 +277,7 @@ func newMirrorStatusCmd(env cliEnv) cobra.Command {
277	277
278	278		func newMirrorSyncCmd(env cliEnv) cobra.Command {
279	279		var all, force bool
280		-	cmd := &cobra.Command{Use: "sync <repo>\|--all", Short: "Sync mirrors now", Long: "Sync one repository, or every repository with --all (owner only).\n--force never force-pushes: it only skips the no-op check and throttle and lifts a pause.", Args: func(_ *cobra.Command, args []string) error {
	280	+	cmd := &cobra.Command{Use: "sync <repo>\|--all", Short: "Sync mirrors now", Long: "Sync one repository, or every repository with --all (owner only).\nRepositories are pushed concurrently. The command waits about 8 s; a push still running then is reported as queued,\nkeeps running on the server, and its outcome is shown by `ermite mirror status`.\n--force never force-pushes: it only skips the no-op check and throttle and lifts a pause.", Args: func(_ *cobra.Command, args []string) error {
281	281		if (all && len(args) == 0) \|\| (!all && len(args) == 1) {
282	282		return nil
283	283		}

@@ -298,9 +298,12 @@ func newMirrorSyncCmd(env cliEnv) cobra.Command {
298	298		if err != nil {
299	299		return c.Describe(err)
300	300		}
301		-	failed := 0
	301	+	failed, pending := 0, 0
302	302		rows := make([][]string, 0, len(res.Msg.GetResults()))
303	303		for _, r := range res.Msg.GetResults() {
	304	+	if r.GetQueued() {
	305	+	pending++
	306	+	}
304	307		reason := r.GetStatus().GetReason()
305	308		if r.GetError() != "" {
306	309		failed++

@@ -311,6 +314,9 @@ func newMirrorSyncCmd(env cliEnv) cobra.Command {
311	314		if err := table(cmd.OutOrStdout(), []string{"REPO", "STATE", "QUEUED", "NO-OP", "REASON"}, rows); err != nil {
312	315		return err
313	316		}
	317	+	if pending > 0 {
	318	+	_, _ = fmt.Fprintf(cmd.ErrOrStderr(), "%d of %d mirror syncs still running: they continue in the background; check `ermite mirror status`\n", pending, len(rows))
	319	+	}
314	320		if failed > 0 {
315	321		return fmt.Errorf("%d of %d mirror syncs failed", failed, len(rows))
316	322		}

 func newMirrorSyncCmd(env *cliEnv) *cobra.Command {
 	var all, force bool
-	cmd := &cobra.Command{Use: "sync <repo>|--all", Short: "Sync mirrors now", Long: "Sync one repository, or every repository with --all (owner only).\n--force never force-pushes: it only skips the no-op check and throttle and lifts a pause.", Args: func(_ *cobra.Command, args []string) error {
+	cmd := &cobra.Command{Use: "sync <repo>|--all", Short: "Sync mirrors now", Long: "Sync one repository, or every repository with --all (owner only).\nRepositories are pushed concurrently. The command waits about 8 s; a push still running then is reported as queued,\nkeeps running on the server, and its outcome is shown by `ermite mirror status`.\n--force never force-pushes: it only skips the no-op check and throttle and lifts a pause.", Args: func(_ *cobra.Command, args []string) error {
 		if (all && len(args) == 0) || (!all && len(args) == 1) {
 			return nil
 		}
 		if err != nil {
 			return c.Describe(err)
 		}
-		failed := 0
+		failed, pending := 0, 0
 		rows := make([][]string, 0, len(res.Msg.GetResults()))
 		for _, r := range res.Msg.GetResults() {
+			if r.GetQueued() {
+				pending++
+			}
 			reason := r.GetStatus().GetReason()
 			if r.GetError() != "" {
 				failed++
 		if err := table(cmd.OutOrStdout(), []string{"REPO", "STATE", "QUEUED", "NO-OP", "REASON"}, rows); err != nil {
 			return err
 		}
+		if pending > 0 {
+			_, _ = fmt.Fprintf(cmd.ErrOrStderr(), "%d of %d mirror syncs still running: they continue in the background; check `ermite mirror status`\n", pending, len(rows))
+		}
 		if failed > 0 {
 			return fmt.Errorf("%d of %d mirror syncs failed", failed, len(rows))
 		}

M internal/app/cli_mirror_test.go +50 -0


@@ -3,11 +3,18 @@
3	3		package app
4	4
5	5		import (
	6	+	"context"
6	7		"strings"
7	8		"testing"
	9	+	"time"
8	10
9	11		"github.com/stretchr/testify/assert"
10	12		"github.com/stretchr/testify/require"
	13	+
	14	+	"bnema.dev/ermite/internal/config"
	15	+	"bnema.dev/ermite/internal/domain"
	16	+	"bnema.dev/ermite/internal/ports"
	17	+	"bnema.dev/ermite/internal/usecase"
11	18		)
12	19
13	20		func TestMirrorCLIArguments(t *testing.T) {

@@ -65,3 +72,46 @@ func TestMirrorCLIConfiguration(t *testing.T) {
65	72		assert.Equal(t, rotated.stdout, run("key", "show").ok(t).stdout)
66	73		assert.Equal(t, []string{"authorized_key", "created_at", "fingerprint"}, keys(t, run("key", "show", "--json").ok(t).stdout))
67	74		}
	75	+
	76	+	func TestMirrorCLISyncHelpDocumentsBackgroundSync(t *testing.T) {
	77	+	r := runCLI(t, nil, "", "mirror", "sync", "--help").ok(t)
	78	+	assert.Contains(t, r.stdout, "queued")
	79	+	assert.Contains(t, r.stdout, "ermite mirror status")
	80	+	}
	81	+
	82	+	type blockingPusher struct{ entered, release chan struct{} }
	83	+
	84	+	func (p blockingPusher) Push(ctx context.Context, _ ports.MirrorPush) (ports.MirrorPushResult, error) {
	85	+	p.entered <- struct{}{}
	86	+	select {
	87	+	case <-p.release:
	88	+	case <-ctx.Done():
	89	+	}
	90	+	return ports.MirrorPushResult{}, nil
	91	+	}
	92	+
	93	+	// A push that outlasts the server's wait bound answers "queued"; the CLI says it continues.
	94	+	func TestMirrorCLISyncReportsBackgroundSync(t *testing.T) {
	95	+	p := blockingPusher{entered: make(chan struct{}, 4), release: make(chan struct{})}
	96	+	h := newHarnessWith(t, harnessOpts{env: map[string]string{config.EnvSecretsKey: secretsKeyA}}, func(o *options) {
	97	+	o.mirrors = func(c *usecase.MirrorsConfig) {
	98	+	c.ValidateURL = func(string) error { return nil }
	99	+	c.Pusher = p
	100	+	c.SyncWait = 50 * time.Millisecond
	101	+	}
	102	+	})
	103	+	h.createRepo("proj", domain.VisibilityPrivate)
	104	+	c := &cliHarness{harness: h, url: h.srv.URL}
	105	+	wt := h.workTree("main", "README.md")
	106	+	h.mustGit(wt, "push", h.urlAuth(adminToken, "/proj.git"), "main")
	107	+	c.cli(t, adminToken, "mirror", "add", "proj", "file:///nowhere", "--cron", "/5 * * *").ok(t)
	108	+	for _, args := range [][]string{{"sync", "proj", "--force"}, {"sync", "--all", "--force"}} {
	109	+	r := c.cli(t, adminToken, append([]string{"mirror"}, args...)...).ok(t)
	110	+	assert.Contains(t, r.stdout, "proj")
	111	+	assert.Contains(t, r.stdout, "true", "QUEUED column")
	112	+	assert.Contains(t, r.stderr, "continue in the background")
	113	+	assert.Contains(t, r.stderr, "ermite mirror status")
	114	+	}
	115	+	close(p.release)
	116	+	h.forge.Mirrors.Wait(h.ctx)
	117	+	}


@@ -3,11 +3,18 @@
3	package app	3		package app
4		4
5	import (	5		import (
		6	+	"context"
6	"strings"	7		"strings"
7	"testing"	8		"testing"
		9	+	"time"
8		10
9	"github.com/stretchr/testify/assert"	11		"github.com/stretchr/testify/assert"
10	"github.com/stretchr/testify/require"	12		"github.com/stretchr/testify/require"
		13	+
		14	+	"bnema.dev/ermite/internal/config"
		15	+	"bnema.dev/ermite/internal/domain"
		16	+	"bnema.dev/ermite/internal/ports"
		17	+	"bnema.dev/ermite/internal/usecase"
11	)	18		)
12		19
13	func TestMirrorCLIArguments(t *testing.T) {	20		func TestMirrorCLIArguments(t *testing.T) {

@@ -65,3 +72,46 @@ func TestMirrorCLIConfiguration(t *testing.T) {
65	assert.Equal(t, rotated.stdout, run("key", "show").ok(t).stdout)	72		assert.Equal(t, rotated.stdout, run("key", "show").ok(t).stdout)
66	assert.Equal(t, []string{"authorized_key", "created_at", "fingerprint"}, keys(t, run("key", "show", "--json").ok(t).stdout))	73		assert.Equal(t, []string{"authorized_key", "created_at", "fingerprint"}, keys(t, run("key", "show", "--json").ok(t).stdout))
67	}	74		}
		75	+
		76	+	func TestMirrorCLISyncHelpDocumentsBackgroundSync(t *testing.T) {
		77	+	r := runCLI(t, nil, "", "mirror", "sync", "--help").ok(t)
		78	+	assert.Contains(t, r.stdout, "queued")
		79	+	assert.Contains(t, r.stdout, "ermite mirror status")
		80	+	}
		81	+
		82	+	type blockingPusher struct{ entered, release chan struct{} }
		83	+
		84	+	func (p blockingPusher) Push(ctx context.Context, _ ports.MirrorPush) (ports.MirrorPushResult, error) {
		85	+	p.entered <- struct{}{}
		86	+	select {
		87	+	case <-p.release:
		88	+	case <-ctx.Done():
		89	+	}
		90	+	return ports.MirrorPushResult{}, nil
		91	+	}
		92	+
		93	+	// A push that outlasts the server's wait bound answers "queued"; the CLI says it continues.
		94	+	func TestMirrorCLISyncReportsBackgroundSync(t *testing.T) {
		95	+	p := blockingPusher{entered: make(chan struct{}, 4), release: make(chan struct{})}
		96	+	h := newHarnessWith(t, harnessOpts{env: map[string]string{config.EnvSecretsKey: secretsKeyA}}, func(o *options) {
		97	+	o.mirrors = func(c *usecase.MirrorsConfig) {
		98	+	c.ValidateURL = func(string) error { return nil }
		99	+	c.Pusher = p
		100	+	c.SyncWait = 50 * time.Millisecond
		101	+	}
		102	+	})
		103	+	h.createRepo("proj", domain.VisibilityPrivate)
		104	+	c := &cliHarness{harness: h, url: h.srv.URL}
		105	+	wt := h.workTree("main", "README.md")
		106	+	h.mustGit(wt, "push", h.urlAuth(adminToken, "/proj.git"), "main")
		107	+	c.cli(t, adminToken, "mirror", "add", "proj", "file:///nowhere", "--cron", "/5 * * *").ok(t)
		108	+	for _, args := range [][]string{{"sync", "proj", "--force"}, {"sync", "--all", "--force"}} {
		109	+	r := c.cli(t, adminToken, append([]string{"mirror"}, args...)...).ok(t)
		110	+	assert.Contains(t, r.stdout, "proj")
		111	+	assert.Contains(t, r.stdout, "true", "QUEUED column")
		112	+	assert.Contains(t, r.stderr, "continue in the background")
		113	+	assert.Contains(t, r.stderr, "ermite mirror status")
		114	+	}
		115	+	close(p.release)
		116	+	h.forge.Mirrors.Wait(h.ctx)
		117	+	}

M internal/usecase/mirrors.go +6 -0


@@ -29,6 +29,9 @@ type MirrorsConfig struct {
29	29		ScheduleTick time.Duration
30	30		Workers int
31	31		QueueSize int // jobs waiting for a worker; default 64
	32	+	// SyncWait bounds how long a manual sync (Sync, SyncAll) holds its caller; past it the
	33	+	// result is Queued and the push continues. Default DefaultMirrorSyncWait.
	34	+	SyncWait time.Duration
32	35		// ValidateURL checks a resolved mirror URL; default domain.ValidateMirrorURL. Tests of
33	36		// local transports inject a permissive check.
34	37		ValidateURL func(string) error

@@ -76,6 +79,9 @@ func NewMirrors(cfg MirrorsConfig) *Mirrors {
76	79		if cfg.QueueSize <= 0 {
77	80		cfg.QueueSize = mirrorQueueSize
78	81		}
	82	+	if cfg.SyncWait <= 0 {
	83	+	cfg.SyncWait = DefaultMirrorSyncWait
	84	+	}
79	85		if cfg.ValidateURL == nil {
80	86		cfg.ValidateURL = domain.ValidateMirrorURL
81	87		}


@@ -29,6 +29,9 @@ type MirrorsConfig struct {
29	ScheduleTick time.Duration	29		ScheduleTick time.Duration
30	Workers int	30		Workers int
31	QueueSize int // jobs waiting for a worker; default 64	31		QueueSize int // jobs waiting for a worker; default 64
		32	+	// SyncWait bounds how long a manual sync (Sync, SyncAll) holds its caller; past it the
		33	+	// result is Queued and the push continues. Default DefaultMirrorSyncWait.
		34	+	SyncWait time.Duration
32	// ValidateURL checks a resolved mirror URL; default domain.ValidateMirrorURL. Tests of	35		// ValidateURL checks a resolved mirror URL; default domain.ValidateMirrorURL. Tests of
33	// local transports inject a permissive check.	36		// local transports inject a permissive check.
34	ValidateURL func(string) error	37		ValidateURL func(string) error

@@ -76,6 +79,9 @@ func NewMirrors(cfg MirrorsConfig) *Mirrors {
76	if cfg.QueueSize <= 0 {	79		if cfg.QueueSize <= 0 {
77	cfg.QueueSize = mirrorQueueSize	80		cfg.QueueSize = mirrorQueueSize
78	}	81		}
		82	+	if cfg.SyncWait <= 0 {
		83	+	cfg.SyncWait = DefaultMirrorSyncWait
		84	+	}
79	if cfg.ValidateURL == nil {	85		if cfg.ValidateURL == nil {
80	cfg.ValidateURL = domain.ValidateMirrorURL	86		cfg.ValidateURL = domain.ValidateMirrorURL
81	}	87		}

M internal/usecase/mirrors_sync.go +72 -20


@@ -36,7 +36,13 @@ var (
36	36		_ ports.MirrorWarner = (*Mirrors)(nil)
37	37		)
38	38
39		-	// Sync waits for its own run; overlapping requests coalesce into one sticky rerun.
	39	+	// DefaultMirrorSyncWait is how long a manual sync holds its caller before answering "queued":
	40	+	// under the 10 s the edge proxy allows a response, and well under the client timeout.
	41	+	const DefaultMirrorSyncWait = 8 * time.Second
	42	+
	43	+	// Sync waits for its own run; overlapping requests coalesce into one sticky rerun. When the
	44	+	// push outlasts the wait bound the result is Queued: the sync goes on in the background and
	45	+	// its outcome is in the mirror status.
40	46		func (m *Mirrors) Sync(ctx context.Context, p domain.Principal, repo domain.RepoName, force bool) (ports.MirrorSyncResult, error) {
41	47		if err := m.admin(ctx, p, repo); err != nil {
42	48		return ports.MirrorSyncResult{}, err

@@ -47,22 +53,47 @@ func (m *Mirrors) Sync(ctx context.Context, p domain.Principal, repo domain.Repo
47	53		return ports.MirrorSyncResult{}, err
48	54		}
49	55		if queued {
50		-	return ports.MirrorSyncResult{Repo: repo, Status: domain.MirrorStatus{Repo: repo}, Queued: true}, nil
	56	+	return pendingSync(repo), nil
51	57		}
	58	+	wctx, cancel := context.WithTimeout(ctx, m.cfg.SyncWait)
	59	+	defer cancel()
	60	+	return m.await(ctx, wctx, repo, f)
	61	+	}
	62	+
	63	+	func pendingSync(repo domain.RepoName) ports.MirrorSyncResult {
	64	+	return ports.MirrorSyncResult{Repo: repo, Status: domain.MirrorStatus{Repo: repo}, Queued: true}
	65	+	}
	66	+
	67	+	// await waits for the first run of f. wctx carries the wait bound and ctx the caller's own
	68	+	// cancellation: the bound expiring alone yields a Queued result, not an error.
	69	+	func (m Mirrors) await(ctx, wctx context.Context, repo domain.RepoName, f mirrorFlight) (ports.MirrorSyncResult, error) {
52	70		select {
53	71		case <-m.ctx.Done():
54	72		return ports.MirrorSyncResult{}, m.ctx.Err()
55		-	case <-ctx.Done():
56		-	return ports.MirrorSyncResult{}, ctx.Err()
	73	+	case <-wctx.Done():
	74	+	if err := ctx.Err(); err != nil {
	75	+	return ports.MirrorSyncResult{}, err
	76	+	}
	77	+	select { // a run that finished as the bound expired wins
	78	+	case <-f.done:
	79	+	default:
	80	+	return pendingSync(repo), nil
	81	+	}
57	82		case <-f.done:
58		-	f.result.Repo = repo
59		-	return f.result, f.err
60	83		}
	84	+	m.mu.Lock()
	85	+	res, err := f.result, f.err
	86	+	m.mu.Unlock()
	87	+	res.Repo = repo
	88	+	return res, err
61	89		}
62	90
63		-	// SyncAll syncs every repository whose mirror is enabled, as an owner. A failing repository
64		-	// does not stop the others: its outcome carries the error. The returned error is only for
65		-	// what prevents the whole request (authorization, listing, cancellation).
	91	+	// SyncAll syncs every repository whose mirror is enabled, as an owner. Every repository is
	92	+	// enqueued first, then all outcomes are awaited together under one wait bound, so the total
	93	+	// time is about one push, not one per repository. A failing repository does not stop the
	94	+	// others: its outcome carries the error; one still running at the bound is Queued. The
	95	+	// returned error is only for what prevents the whole request (authorization, listing,
	96	+	// cancellation). Results are in listing order.
66	97		func (m *Mirrors) SyncAll(ctx context.Context, p domain.Principal, force bool) ([]ports.MirrorSyncResult, error) {
67	98		if err := mirrorOwner(p); err != nil {
68	99		return nil, err

@@ -71,28 +102,49 @@ func (m *Mirrors) SyncAll(ctx context.Context, p domain.Principal, force bool) (
71	102		if err != nil {
72	103		return nil, err
73	104		}
74		-	out := make([]ports.MirrorSyncResult, 0, len(repos))
	105	+	m.Start()
	106	+	type entry struct {
	107	+	res ports.MirrorSyncResult
	108	+	f *mirrorFlight // nil once the outcome is known
	109	+	}
	110	+	entries := make([]entry, 0, len(repos))
75	111		for _, r := range repos {
76	112		if err := ctx.Err(); err != nil {
77		-	return out, err
	113	+	return nil, err
78	114		}
79	115		t, err := m.target(ctx, r.Name)
80	116		if err == nil && !t.Enabled {
81	117		continue
82	118		}
83		-	var res ports.MirrorSyncResult
	119	+	e := entry{res: ports.MirrorSyncResult{Repo: r.Name, Err: err}}
84	120		if err == nil {
85		-	res, err = m.Sync(ctx, p, r.Name, force)
	121	+	f, queued, qerr := m.enqueue(r.Name, mirrorReq{force: force, manual: true})
	122	+	switch {
	123	+	case qerr != nil:
	124	+	e.res.Err = qerr
	125	+	case queued:
	126	+	e.res = pendingSync(r.Name)
	127	+	default:
	128	+	e.f = f
	129	+	}
86	130		}
87		-	if ctxErr := ctx.Err(); ctxErr != nil {
88		-	return out, ctxErr
	131	+	entries = append(entries, e)
	132	+	}
	133	+	wctx, cancel := context.WithTimeout(ctx, m.cfg.SyncWait)
	134	+	defer cancel()
	135	+	out := make([]ports.MirrorSyncResult, 0, len(entries))
	136	+	for _, e := range entries {
	137	+	if e.f != nil {
	138	+	res, err := m.await(ctx, wctx, e.res.Repo, e.f)
	139	+	if cerr := ctx.Err(); cerr != nil {
	140	+	return nil, cerr
	141	+	}
	142	+	e.res, e.res.Err = res, err
89	143		}
90		-	res.Repo = r.Name
91		-	if res.Status.Repo == (domain.RepoName{}) {
92		-	res.Status.Repo = r.Name
	144	+	if e.res.Status.Repo == (domain.RepoName{}) {
	145	+	e.res.Status.Repo = e.res.Repo
93	146		}
94		-	res.Err = err
95		-	out = append(out, res)
	147	+	out = append(out, e.res)
96	148		}
97	149		return out, nil
98	150		}

 	_ ports.MirrorWarner  = (*Mirrors)(nil)
 )
-// Sync waits for its own run; overlapping requests coalesce into one sticky rerun.
+// DefaultMirrorSyncWait is how long a manual sync holds its caller before answering "queued":
+// under the 10 s the edge proxy allows a response, and well under the client timeout.
+const DefaultMirrorSyncWait = 8 * time.Second
+// Sync waits for its own run; overlapping requests coalesce into one sticky rerun. When the
+// push outlasts the wait bound the result is Queued: the sync goes on in the background and
+// its outcome is in the mirror status.
 func (m *Mirrors) Sync(ctx context.Context, p domain.Principal, repo domain.RepoName, force bool) (ports.MirrorSyncResult, error) {
 	if err := m.admin(ctx, p, repo); err != nil {
 		return ports.MirrorSyncResult{}, err
 		return ports.MirrorSyncResult{}, err
 	}
 	if queued {
-		return ports.MirrorSyncResult{Repo: repo, Status: domain.MirrorStatus{Repo: repo}, Queued: true}, nil
+		return pendingSync(repo), nil
 	}
+	wctx, cancel := context.WithTimeout(ctx, m.cfg.SyncWait)
+	defer cancel()
+	return m.await(ctx, wctx, repo, f)
+}
+func pendingSync(repo domain.RepoName) ports.MirrorSyncResult {
+	return ports.MirrorSyncResult{Repo: repo, Status: domain.MirrorStatus{Repo: repo}, Queued: true}
+}
+// await waits for the first run of f. wctx carries the wait bound and ctx the caller's own
+// cancellation: the bound expiring alone yields a Queued result, not an error.
+func (m *Mirrors) await(ctx, wctx context.Context, repo domain.RepoName, f *mirrorFlight) (ports.MirrorSyncResult, error) {
 	select {
 	case <-m.ctx.Done():
 		return ports.MirrorSyncResult{}, m.ctx.Err()
-	case <-ctx.Done():
+	case <-wctx.Done():
-		return ports.MirrorSyncResult{}, ctx.Err()
+		if err := ctx.Err(); err != nil {
+			return ports.MirrorSyncResult{}, err
+		}
+		select { // a run that finished as the bound expired wins
+		case <-f.done:
+		default:
+			return pendingSync(repo), nil
+		}
 	case <-f.done:
-		f.result.Repo = repo
-		return f.result, f.err
 	}
+	m.mu.Lock()
+	res, err := f.result, f.err
+	m.mu.Unlock()
+	res.Repo = repo
+	return res, err
 }
-// SyncAll syncs every repository whose mirror is enabled, as an owner. A failing repository
+// SyncAll syncs every repository whose mirror is enabled, as an owner. Every repository is
-// does not stop the others: its outcome carries the error. The returned error is only for
+// enqueued first, then all outcomes are awaited together under one wait bound, so the total
-// what prevents the whole request (authorization, listing, cancellation).
+// time is about one push, not one per repository. A failing repository does not stop the
+// others: its outcome carries the error; one still running at the bound is Queued. The
+// returned error is only for what prevents the whole request (authorization, listing,
+// cancellation). Results are in listing order.
 func (m *Mirrors) SyncAll(ctx context.Context, p domain.Principal, force bool) ([]ports.MirrorSyncResult, error) {
 	if err := mirrorOwner(p); err != nil {
 		return nil, err
 	if err != nil {
 		return nil, err
 	}
-	out := make([]ports.MirrorSyncResult, 0, len(repos))
+	m.Start()
+	type entry struct {
+		res ports.MirrorSyncResult
+		f   *mirrorFlight // nil once the outcome is known
+	}
+	entries := make([]entry, 0, len(repos))
 	for _, r := range repos {
 		if err := ctx.Err(); err != nil {
-			return out, err
+			return nil, err
 		}
 		t, err := m.target(ctx, r.Name)
 		if err == nil && !t.Enabled {
 			continue
 		}
-		var res ports.MirrorSyncResult
+		e := entry{res: ports.MirrorSyncResult{Repo: r.Name, Err: err}}
 		if err == nil {
-			res, err = m.Sync(ctx, p, r.Name, force)
+			f, queued, qerr := m.enqueue(r.Name, mirrorReq{force: force, manual: true})
+			switch {
+			case qerr != nil:
+				e.res.Err = qerr
+			case queued:
+				e.res = pendingSync(r.Name)
+			default:
+				e.f = f
+			}
 		}
-		if ctxErr := ctx.Err(); ctxErr != nil {
+		entries = append(entries, e)
-			return out, ctxErr
+	}
+	wctx, cancel := context.WithTimeout(ctx, m.cfg.SyncWait)
+	defer cancel()
+	out := make([]ports.MirrorSyncResult, 0, len(entries))
+	for _, e := range entries {
+		if e.f != nil {
+			res, err := m.await(ctx, wctx, e.res.Repo, e.f)
+			if cerr := ctx.Err(); cerr != nil {
+				return nil, cerr
+			}
+			e.res, e.res.Err = res, err
 		}
-		res.Repo = r.Name
+		if e.res.Status.Repo == (domain.RepoName{}) {
-		if res.Status.Repo == (domain.RepoName{}) {
+			e.res.Status.Repo = e.res.Repo
-			res.Status.Repo = r.Name
 		}
-		res.Err = err
+		out = append(out, e.res)
-		out = append(out, res)
 	}
 	return out, nil
 }

M internal/usecase/mirrors_sync_test.go +90 -0


@@ -490,3 +490,93 @@ func TestMirrorsAuthorizationAndMissingSecretsKey(t *testing.T) {
490	490		_, err = f.m.RotateKey(context.Background(), tokenPrincipal(domain.ScopeRepoRead))
491	491		require.ErrorIs(t, err, domain.ErrForbidden)
492	492		}
	493	+
	494	+	// gatedPush makes every push announce itself on entered and block until release is closed.
	495	+	func (f *mirrorFixture) gatedPush(release <-chan struct{}) <-chan struct{} {
	496	+	entered := make(chan struct{}, 16)
	497	+	f.push.EXPECT().Push(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, _ ports.MirrorPush) (ports.MirrorPushResult, error) {
	498	+	entered <- struct{}{}
	499	+	select {
	500	+	case <-release:
	501	+	case <-ctx.Done():
	502	+	}
	503	+	return ports.MirrorPushResult{}, nil
	504	+	}).Maybe()
	505	+	return entered
	506	+	}
	507	+
	508	+	func TestMirrorsSyncAllRunsRepositoriesConcurrently(t *testing.T) {
	509	+	f := newMirrorFixtureWith(t, func(c *usecase.MirrorsConfig) { c.Workers = 3; c.SyncWait = time.Minute })
	510	+	f.addRepo(name(t, "b"))
	511	+	f.addRepo(name(t, "c"))
	512	+	release := make(chan struct{})
	513	+	entered := f.gatedPush(release)
	514	+	type outcome struct {
	515	+	got []ports.MirrorSyncResult
	516	+	err error
	517	+	}
	518	+	done := make(chan outcome, 1)
	519	+	go func() {
	520	+	got, err := f.m.SyncAll(context.Background(), domain.Root(), true)
	521	+	done <- outcome{got, err}
	522	+	}()
	523	+	// All three pushes are in flight together: a sequential SyncAll would never get past the first.
	524	+	for i := 0; i < 3; i++ {
	525	+	select {
	526	+	case <-entered:
	527	+	case <-time.After(10 * time.Second):
	528	+	t.Fatalf("only %d pushes started: repositories are not synced concurrently", i)
	529	+	}
	530	+	}
	531	+	close(release)
	532	+	o := <-done
	533	+	require.NoError(t, o.err)
	534	+	require.Len(t, o.got, 3)
	535	+	for i, want := range []string{"tools", "b", "c"} {
	536	+	r := o.got[i]
	537	+	require.Equal(t, want, r.Repo.String(), "results keep the listing order")
	538	+	require.NoError(t, r.Err)
	539	+	require.False(t, r.Queued)
	540	+	require.Equal(t, domain.MirrorOK, r.Status.State)
	541	+	}
	542	+	}
	543	+
	544	+	func TestMirrorsSyncReturnsQueuedWhenPushOutlastsTheWait(t *testing.T) {
	545	+	f := newMirrorFixtureWith(t, func(c usecase.MirrorsConfig) { c.SyncWait = 20 time.Millisecond })
	546	+	other := name(t, "other")
	547	+	f.addRepo(other)
	548	+	release := make(chan struct{})
	549	+	entered := f.gatedPush(release)
	550	+
	551	+	res, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
	552	+	require.NoError(t, err)
	553	+	require.True(t, res.Queued, "the push is still running at the bound")
	554	+	require.Equal(t, f.repo, res.Repo)
	555	+	<-entered
	556	+
	557	+	all, err := f.m.SyncAll(context.Background(), domain.Root(), true)
	558	+	require.NoError(t, err)
	559	+	require.Len(t, all, 2)
	560	+	for _, r := range all {
	561	+	require.NoError(t, r.Err)
	562	+	require.True(t, r.Queued, "%s is coalesced or still running", r.Repo)
	563	+	require.Equal(t, r.Repo, r.Status.Repo)
	564	+	}
	565	+
	566	+	close(release)
	567	+	f.drain(t)
	568	+	require.Equal(t, domain.MirrorOK, f.statusOf(f.repo).State, "the sync completed in the background")
	569	+	}
	570	+
	571	+	func TestMirrorsSyncAllHonoursCallerCancellation(t *testing.T) {
	572	+	f := newMirrorFixtureWith(t, func(c *usecase.MirrorsConfig) { c.SyncWait = time.Minute })
	573	+	release := make(chan struct{})
	574	+	entered := f.gatedPush(release)
	575	+	ctx, cancel := context.WithCancel(context.Background())
	576	+	errc := make(chan error, 1)
	577	+	go func() { _, err := f.m.SyncAll(ctx, domain.Root(), true); errc <- err }()
	578	+	<-entered
	579	+	cancel()
	580	+	require.ErrorIs(t, <-errc, context.Canceled)
	581	+	close(release)
	582	+	}


@@ -490,3 +490,93 @@ func TestMirrorsAuthorizationAndMissingSecretsKey(t *testing.T) {
490	_, err = f.m.RotateKey(context.Background(), tokenPrincipal(domain.ScopeRepoRead))	490		_, err = f.m.RotateKey(context.Background(), tokenPrincipal(domain.ScopeRepoRead))
491	require.ErrorIs(t, err, domain.ErrForbidden)	491		require.ErrorIs(t, err, domain.ErrForbidden)
492	}	492		}
		493	+
		494	+	// gatedPush makes every push announce itself on entered and block until release is closed.
		495	+	func (f *mirrorFixture) gatedPush(release <-chan struct{}) <-chan struct{} {
		496	+	entered := make(chan struct{}, 16)
		497	+	f.push.EXPECT().Push(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, _ ports.MirrorPush) (ports.MirrorPushResult, error) {
		498	+	entered <- struct{}{}
		499	+	select {
		500	+	case <-release:
		501	+	case <-ctx.Done():
		502	+	}
		503	+	return ports.MirrorPushResult{}, nil
		504	+	}).Maybe()
		505	+	return entered
		506	+	}
		507	+
		508	+	func TestMirrorsSyncAllRunsRepositoriesConcurrently(t *testing.T) {
		509	+	f := newMirrorFixtureWith(t, func(c *usecase.MirrorsConfig) { c.Workers = 3; c.SyncWait = time.Minute })
		510	+	f.addRepo(name(t, "b"))
		511	+	f.addRepo(name(t, "c"))
		512	+	release := make(chan struct{})
		513	+	entered := f.gatedPush(release)
		514	+	type outcome struct {
		515	+	got []ports.MirrorSyncResult
		516	+	err error
		517	+	}
		518	+	done := make(chan outcome, 1)
		519	+	go func() {
		520	+	got, err := f.m.SyncAll(context.Background(), domain.Root(), true)
		521	+	done <- outcome{got, err}
		522	+	}()
		523	+	// All three pushes are in flight together: a sequential SyncAll would never get past the first.
		524	+	for i := 0; i < 3; i++ {
		525	+	select {
		526	+	case <-entered:
		527	+	case <-time.After(10 * time.Second):
		528	+	t.Fatalf("only %d pushes started: repositories are not synced concurrently", i)
		529	+	}
		530	+	}
		531	+	close(release)
		532	+	o := <-done
		533	+	require.NoError(t, o.err)
		534	+	require.Len(t, o.got, 3)
		535	+	for i, want := range []string{"tools", "b", "c"} {
		536	+	r := o.got[i]
		537	+	require.Equal(t, want, r.Repo.String(), "results keep the listing order")
		538	+	require.NoError(t, r.Err)
		539	+	require.False(t, r.Queued)
		540	+	require.Equal(t, domain.MirrorOK, r.Status.State)
		541	+	}
		542	+	}
		543	+
		544	+	func TestMirrorsSyncReturnsQueuedWhenPushOutlastsTheWait(t *testing.T) {
		545	+	f := newMirrorFixtureWith(t, func(c usecase.MirrorsConfig) { c.SyncWait = 20 time.Millisecond })
		546	+	other := name(t, "other")
		547	+	f.addRepo(other)
		548	+	release := make(chan struct{})
		549	+	entered := f.gatedPush(release)
		550	+
		551	+	res, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
		552	+	require.NoError(t, err)
		553	+	require.True(t, res.Queued, "the push is still running at the bound")
		554	+	require.Equal(t, f.repo, res.Repo)
		555	+	<-entered
		556	+
		557	+	all, err := f.m.SyncAll(context.Background(), domain.Root(), true)
		558	+	require.NoError(t, err)
		559	+	require.Len(t, all, 2)
		560	+	for _, r := range all {
		561	+	require.NoError(t, r.Err)
		562	+	require.True(t, r.Queued, "%s is coalesced or still running", r.Repo)
		563	+	require.Equal(t, r.Repo, r.Status.Repo)
		564	+	}
		565	+
		566	+	close(release)
		567	+	f.drain(t)
		568	+	require.Equal(t, domain.MirrorOK, f.statusOf(f.repo).State, "the sync completed in the background")
		569	+	}
		570	+
		571	+	func TestMirrorsSyncAllHonoursCallerCancellation(t *testing.T) {
		572	+	f := newMirrorFixtureWith(t, func(c *usecase.MirrorsConfig) { c.SyncWait = time.Minute })
		573	+	release := make(chan struct{})
		574	+	entered := f.gatedPush(release)
		575	+	ctx, cancel := context.WithCancel(context.Background())
		576	+	errc := make(chan error, 1)
		577	+	go func() { _, err := f.m.SyncAll(ctx, domain.Root(), true); errc <- err }()
		578	+	<-entered
		579	+	cancel()
		580	+	require.ErrorIs(t, <-errc, context.Canceled)
		581	+	close(release)
		582	+	}