feat(mirror): add the git push transport and the mirror KV storage

commit: e3ee891afa90f7752317b24a4c55e1dd7110e6c1
author: brice · 2d ago · 2026-10-20 05:35 UTC
committer: brice · 2d ago · 2026-10-20 10:02 UTC
parent: 1069a88
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

8 files · +1406 -0

A gen/ermite/storage/mirror.pb.go +577 -0 generated file: collapsed

A internal/adapters/gitpush/gitpush.go +229 -0

@@ -0,0 +1,229 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	// Package gitpush implements bounded, non-forcing mirror pushes over git/ssh.
	4	+	package gitpush
	5	+
	6	+	import (
	7	+	"bytes"
	8	+	"context"
	9	+	"crypto/ed25519"
	10	+	"encoding/pem"
	11	+	"errors"
	12	+	"os"
	13	+	"os/exec"
	14	+	"strings"
	15	+	"time"
	16	+	"unicode"
	17	+
	18	+	"bnema.dev/ermite/internal/domain"
	19	+	"bnema.dev/ermite/internal/ports"
	20	+	"golang.org/x/crypto/ssh"
	21	+	)
	22	+
	23	+	type Config struct {
	24	+	Bin, Path, Home, KnownHosts, TempDir string
	25	+	Timeout time.Duration
	26	+	MaxOutput int
	27	+	AllowedProtocols string
	28	+	SSHOptions []string
	29	+	RepoDir func(domain.RepoName) string
	30	+	}
	31	+	type Pusher struct{ cfg Config }
	32	+
	33	+	var _ ports.MirrorPusher = (*Pusher)(nil)
	34	+
	35	+	func New(cfg Config) *Pusher {
	36	+	if cfg.Timeout <= 0 {
	37	+	cfg.Timeout = 10 * time.Minute
	38	+	}
	39	+	if cfg.MaxOutput <= 0 {
	40	+	cfg.MaxOutput = 1 << 20
	41	+	}
	42	+	if cfg.AllowedProtocols == "" {
	43	+	cfg.AllowedProtocols = "ssh"
	44	+	}
	45	+	cfg.SSHOptions = append([]string(nil), cfg.SSHOptions...)
	46	+	return &Pusher{cfg: cfg}
	47	+	}
	48	+	func quote(s string) string { return "'" + strings.ReplaceAll(s, "'", "'\"'\"'") + "'" }
	49	+
	50	+	// validRefspec deliberately accepts only explicit same-name heads and tags.
	51	+	// Empty sources, wildcards, force prefixes and revision expressions never reach git.
	52	+	func validRefspec(s string) bool {
	53	+	a, b, ok := strings.Cut(s, ":")
	54	+	if !ok \|\| a != b \|\| !(strings.HasPrefix(a, "refs/heads/") \|\| strings.HasPrefix(a, "refs/tags/")) {
	55	+	return false
	56	+	}
	57	+	if strings.ContainsAny(a, "+*?[:\\^~ ") \|\| strings.Contains(a, "..") \|\| strings.Contains(a, "@{") \|\| strings.Contains(a, "//") \|\| strings.HasSuffix(a, "/") \|\| strings.HasSuffix(a, ".") {
	58	+	return false
	59	+	}
	60	+	for _, part := range strings.Split(a, "/") {
	61	+	if part == "" \|\| strings.HasPrefix(part, ".") \|\| strings.HasSuffix(part, ".lock") {
	62	+	return false
	63	+	}
	64	+	}
	65	+	return !strings.ContainsFunc(a, unicode.IsControl)
	66	+	}
	67	+
	68	+	func (p *Pusher) Push(parent context.Context, req ports.MirrorPush) (result ports.MirrorPushResult, err error) {
	69	+	start := time.Now()
	70	+	defer func() { result.Duration = time.Since(start) }()
	71	+	internal := func(detail string) error {
	72	+	return &domain.MirrorError{Class: domain.MirrorClassInternal, Detail: detail}
	73	+	}
	74	+	if err := parent.Err(); err != nil {
	75	+	return result, err
	76	+	}
	77	+	if len(req.PrivateKey) != ed25519.PrivateKeySize {
	78	+	return result, internal("invalid mirror private key")
	79	+	}
	80	+	if p.cfg.RepoDir == nil \|\| p.cfg.TempDir == "" \|\| p.cfg.Home == "" \|\| p.cfg.KnownHosts == "" {
	81	+	return result, internal("mirror transport directories are not configured")
	82	+	}
	83	+	if req.URL == "" \|\| strings.HasPrefix(req.URL, "-") \|\| strings.ContainsFunc(req.URL, unicode.IsControl) {
	84	+	return result, internal("invalid mirror target")
	85	+	}
	86	+	for _, ref := range req.Refspecs {
	87	+	if !validRefspec(ref) {
	88	+	return result, internal("invalid mirror refspec")
	89	+	}
	90	+	}
	91	+	if len(req.Refspecs) == 0 {
	92	+	return result, nil
	93	+	} // never let git select implicit refspecs
	94	+	ctx, cancel := context.WithTimeout(parent, p.cfg.Timeout)
	95	+	defer cancel()
	96	+	if err := os.MkdirAll(p.cfg.TempDir, 0700); err != nil {
	97	+	return result, internal("create mirror key directory")
	98	+	}
	99	+	if err := os.Chmod(p.cfg.TempDir, 0700); err != nil {
	100	+	return result, internal("protect mirror key directory")
	101	+	}
	102	+	dir, e := os.MkdirTemp(p.cfg.TempDir, "push-")
	103	+	if e != nil {
	104	+	return result, internal("create temporary mirror directory")
	105	+	}
	106	+	defer os.RemoveAll(dir)
	107	+	block, e := ssh.MarshalPrivateKey(req.PrivateKey, "")
	108	+	if e != nil {
	109	+	return result, internal("encode mirror private key")
	110	+	}
	111	+	f, e := os.CreateTemp(dir, "key-")
	112	+	if e != nil {
	113	+	return result, internal("create temporary mirror key")
	114	+	}
	115	+	keyPath := f.Name()
	116	+	_, e = f.Write(pem.EncodeToMemory(block))
	117	+	closeErr := f.Close()
	118	+	if e != nil \|\| closeErr != nil {
	119	+	return result, internal("write temporary mirror key")
	120	+	}
	121	+	// -F /dev/null prevents ~/.ssh/config and system config from adding credentials
	122	+	// or weakening the required options. Required options precede caller extras.
	123	+	command := "ssh -F /dev/null -i " + quote(keyPath)
	124	+	for _, opt := range append([]string{"IdentitiesOnly=yes", "BatchMode=yes", "StrictHostKeyChecking=accept-new", "UserKnownHostsFile=" + p.cfg.KnownHosts}, p.cfg.SSHOptions...) {
	125	+	command += " -o " + quote(opt)
	126	+	}
	127	+	cleanup := max(min(2*time.Second, p.cfg.Timeout/8), time.Millisecond)
	128	+	runCtx, stop := context.WithTimeout(ctx, max(p.cfg.Timeout-cleanup, time.Millisecond))
	129	+	defer stop()
	130	+	stdout := &limitWriter{max: p.cfg.MaxOutput, cancel: stop}
	131	+	stderr := &limitWriter{max: p.cfg.MaxOutput, cancel: stop}
	132	+	args := []string{"-C", p.cfg.RepoDir(req.Repo), "push", "--porcelain", "--no-verify", "--atomic", "--", req.URL}
	133	+	args = append(args, req.Refspecs...)
	134	+	cmd := exec.Command(p.cfg.Bin, args...)
	135	+	cmd.Env = []string{"PATH=" + p.cfg.Path, "HOME=" + p.cfg.Home, "GIT_ALLOW_PROTOCOL=" + p.cfg.AllowedProtocols, "GIT_TERMINAL_PROMPT=0", "GIT_CONFIG_NOSYSTEM=1", "GIT_SSH_COMMAND=" + command, "LC_ALL=C"}
	136	+	cmd.Dir = p.cfg.Home
	137	+	cmd.Stdout = stdout
	138	+	cmd.Stderr = stderr
	139	+	cmd.WaitDelay = cleanup
	140	+	runErr := run(runCtx, cmd)
	141	+	if parent.Err() != nil {
	142	+	return result, parent.Err()
	143	+	}
	144	+	rejected := false
	145	+	for line := range strings.SplitSeq(stdout.buf.String(), "\n") {
	146	+	fields := strings.Split(line, "\t")
	147	+	if len(fields) < 3 \|\| len(fields[0]) != 1 {
	148	+	continue
	149	+	}
	150	+	var status ports.MirrorRefStatus
	151	+	switch fields[0] {
	152	+	case " ", "*":
	153	+	status = ports.MirrorRefOK
	154	+	case "=":
	155	+	status = ports.MirrorRefUpToDate
	156	+	case "!":
	157	+	status = ports.MirrorRefRejected
	158	+	rejected = true
	159	+	default:
	160	+	continue
	161	+	}
	162	+	_, ref, _ := strings.Cut(fields[1], ":")
	163	+	result.Refs = append(result.Refs, ports.MirrorRefOutcome{Ref: ref, Status: status, Reason: safeDetail(fields[2], req.URL)})
	164	+	}
	165	+	if stdout.exceeded \|\| stderr.exceeded {
	166	+	return result, &domain.MirrorError{Class: domain.MirrorClassTransient, Detail: "mirror output limit exceeded"}
	167	+	}
	168	+	if runCtx.Err() != nil {
	169	+	return result, &domain.MirrorError{Class: domain.MirrorClassTransient, Detail: "mirror push timed out"}
	170	+	}
	171	+	if runErr != nil \|\| rejected {
	172	+	class := domain.ClassifyMirrorOutput(stderr.buf.String()+"\n"+stdout.buf.String(), runErr, false)
	173	+	if rejected && class == domain.MirrorClassNone {
	174	+	class = domain.MirrorClassDiverged
	175	+	}
	176	+	var execErr *exec.Error
	177	+	if errors.As(runErr, &execErr) \|\| localGitError(stderr.buf.String()) {
	178	+	class = domain.MirrorClassInternal
	179	+	}
	180	+	detail := safeDetail(stderr.buf.String(), req.URL)
	181	+	if detail == "" {
	182	+	detail = "git push failed"
	183	+	}
	184	+	return result, &domain.MirrorError{Class: class, Detail: detail}
	185	+	}
	186	+	return result, nil
	187	+	}
	188	+
	189	+	func localGitError(stderr string) bool {
	190	+	for _, marker := range []string{"cannot change to", "not a git repository", "src refspec", "bad object", "unable to read"} {
	191	+	if strings.Contains(strings.ToLower(stderr), marker) {
	192	+	return true
	193	+	}
	194	+	}
	195	+	return false
	196	+	}
	197	+
	198	+	// safeDetail removes the complete target before taking a bounded stderr tail.
	199	+	// SSH diagnostics may repeat the user separately, so suppress user@ tokens too.
	200	+	func safeDetail(s, url string) string {
	201	+	s = strings.ReplaceAll(s, url, "[mirror target]")
	202	+	words := strings.Fields(s)
	203	+	for i, w := range words {
	204	+	if strings.Contains(w, "@") \|\| strings.Contains(w, "://") {
	205	+	words[i] = "[redacted]"
	206	+	}
	207	+	}
	208	+	s = strings.Join(words, " ")
	209	+	if len(s) > 4096 {
	210	+	s = s[len(s)-4096:]
	211	+	}
	212	+	return s
	213	+	}
	214	+
	215	+	type limitWriter struct {
	216	+	buf bytes.Buffer
	217	+	max int
	218	+	exceeded bool
	219	+	cancel context.CancelFunc
	220	+	}
	221	+
	222	+	func (w *limitWriter) Write(p []byte) (int, error) {
	223	+	if len(p) > w.max-w.buf.Len() {
	224	+	w.exceeded = true
	225	+	w.cancel()
	226	+	return 0, errors.New("output limit reached")
	227	+	}
	228	+	return w.buf.Write(p)
	229	+	}

@@ -0,0 +1,229 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	// Package gitpush implements bounded, non-forcing mirror pushes over git/ssh.
	4	+	package gitpush
	5	+
	6	+	import (
	7	+	"bytes"
	8	+	"context"
	9	+	"crypto/ed25519"
	10	+	"encoding/pem"
	11	+	"errors"
	12	+	"os"
	13	+	"os/exec"
	14	+	"strings"
	15	+	"time"
	16	+	"unicode"
	17	+
	18	+	"bnema.dev/ermite/internal/domain"
	19	+	"bnema.dev/ermite/internal/ports"
	20	+	"golang.org/x/crypto/ssh"
	21	+	)
	22	+
	23	+	type Config struct {
	24	+	Bin, Path, Home, KnownHosts, TempDir string
	25	+	Timeout time.Duration
	26	+	MaxOutput int
	27	+	AllowedProtocols string
	28	+	SSHOptions []string
	29	+	RepoDir func(domain.RepoName) string
	30	+	}
	31	+	type Pusher struct{ cfg Config }
	32	+
	33	+	var _ ports.MirrorPusher = (*Pusher)(nil)
	34	+
	35	+	func New(cfg Config) *Pusher {
	36	+	if cfg.Timeout <= 0 {
	37	+	cfg.Timeout = 10 * time.Minute
	38	+	}
	39	+	if cfg.MaxOutput <= 0 {
	40	+	cfg.MaxOutput = 1 << 20
	41	+	}
	42	+	if cfg.AllowedProtocols == "" {
	43	+	cfg.AllowedProtocols = "ssh"
	44	+	}
	45	+	cfg.SSHOptions = append([]string(nil), cfg.SSHOptions...)
	46	+	return &Pusher{cfg: cfg}
	47	+	}
	48	+	func quote(s string) string { return "'" + strings.ReplaceAll(s, "'", "'\"'\"'") + "'" }
	49	+
	50	+	// validRefspec deliberately accepts only explicit same-name heads and tags.
	51	+	// Empty sources, wildcards, force prefixes and revision expressions never reach git.
	52	+	func validRefspec(s string) bool {
	53	+	a, b, ok := strings.Cut(s, ":")
	54	+	if !ok \|\| a != b \|\| !(strings.HasPrefix(a, "refs/heads/") \|\| strings.HasPrefix(a, "refs/tags/")) {
	55	+	return false
	56	+	}
	57	+	if strings.ContainsAny(a, "+*?[:\\^~ ") \|\| strings.Contains(a, "..") \|\| strings.Contains(a, "@{") \|\| strings.Contains(a, "//") \|\| strings.HasSuffix(a, "/") \|\| strings.HasSuffix(a, ".") {
	58	+	return false
	59	+	}
	60	+	for _, part := range strings.Split(a, "/") {
	61	+	if part == "" \|\| strings.HasPrefix(part, ".") \|\| strings.HasSuffix(part, ".lock") {
	62	+	return false
	63	+	}
	64	+	}
	65	+	return !strings.ContainsFunc(a, unicode.IsControl)
	66	+	}
	67	+
	68	+	func (p *Pusher) Push(parent context.Context, req ports.MirrorPush) (result ports.MirrorPushResult, err error) {
	69	+	start := time.Now()
	70	+	defer func() { result.Duration = time.Since(start) }()
	71	+	internal := func(detail string) error {
	72	+	return &domain.MirrorError{Class: domain.MirrorClassInternal, Detail: detail}
	73	+	}
	74	+	if err := parent.Err(); err != nil {
	75	+	return result, err
	76	+	}
	77	+	if len(req.PrivateKey) != ed25519.PrivateKeySize {
	78	+	return result, internal("invalid mirror private key")
	79	+	}
	80	+	if p.cfg.RepoDir == nil \|\| p.cfg.TempDir == "" \|\| p.cfg.Home == "" \|\| p.cfg.KnownHosts == "" {
	81	+	return result, internal("mirror transport directories are not configured")
	82	+	}
	83	+	if req.URL == "" \|\| strings.HasPrefix(req.URL, "-") \|\| strings.ContainsFunc(req.URL, unicode.IsControl) {
	84	+	return result, internal("invalid mirror target")
	85	+	}
	86	+	for _, ref := range req.Refspecs {
	87	+	if !validRefspec(ref) {
	88	+	return result, internal("invalid mirror refspec")
	89	+	}
	90	+	}
	91	+	if len(req.Refspecs) == 0 {
	92	+	return result, nil
	93	+	} // never let git select implicit refspecs
	94	+	ctx, cancel := context.WithTimeout(parent, p.cfg.Timeout)
	95	+	defer cancel()
	96	+	if err := os.MkdirAll(p.cfg.TempDir, 0700); err != nil {
	97	+	return result, internal("create mirror key directory")
	98	+	}
	99	+	if err := os.Chmod(p.cfg.TempDir, 0700); err != nil {
	100	+	return result, internal("protect mirror key directory")
	101	+	}
	102	+	dir, e := os.MkdirTemp(p.cfg.TempDir, "push-")
	103	+	if e != nil {
	104	+	return result, internal("create temporary mirror directory")
	105	+	}
	106	+	defer os.RemoveAll(dir)
	107	+	block, e := ssh.MarshalPrivateKey(req.PrivateKey, "")
	108	+	if e != nil {
	109	+	return result, internal("encode mirror private key")
	110	+	}
	111	+	f, e := os.CreateTemp(dir, "key-")
	112	+	if e != nil {
	113	+	return result, internal("create temporary mirror key")
	114	+	}
	115	+	keyPath := f.Name()
	116	+	_, e = f.Write(pem.EncodeToMemory(block))
	117	+	closeErr := f.Close()
	118	+	if e != nil \|\| closeErr != nil {
	119	+	return result, internal("write temporary mirror key")
	120	+	}
	121	+	// -F /dev/null prevents ~/.ssh/config and system config from adding credentials
	122	+	// or weakening the required options. Required options precede caller extras.
	123	+	command := "ssh -F /dev/null -i " + quote(keyPath)
	124	+	for _, opt := range append([]string{"IdentitiesOnly=yes", "BatchMode=yes", "StrictHostKeyChecking=accept-new", "UserKnownHostsFile=" + p.cfg.KnownHosts}, p.cfg.SSHOptions...) {
	125	+	command += " -o " + quote(opt)
	126	+	}
	127	+	cleanup := max(min(2*time.Second, p.cfg.Timeout/8), time.Millisecond)
	128	+	runCtx, stop := context.WithTimeout(ctx, max(p.cfg.Timeout-cleanup, time.Millisecond))
	129	+	defer stop()
	130	+	stdout := &limitWriter{max: p.cfg.MaxOutput, cancel: stop}
	131	+	stderr := &limitWriter{max: p.cfg.MaxOutput, cancel: stop}
	132	+	args := []string{"-C", p.cfg.RepoDir(req.Repo), "push", "--porcelain", "--no-verify", "--atomic", "--", req.URL}
	133	+	args = append(args, req.Refspecs...)
	134	+	cmd := exec.Command(p.cfg.Bin, args...)
	135	+	cmd.Env = []string{"PATH=" + p.cfg.Path, "HOME=" + p.cfg.Home, "GIT_ALLOW_PROTOCOL=" + p.cfg.AllowedProtocols, "GIT_TERMINAL_PROMPT=0", "GIT_CONFIG_NOSYSTEM=1", "GIT_SSH_COMMAND=" + command, "LC_ALL=C"}
	136	+	cmd.Dir = p.cfg.Home
	137	+	cmd.Stdout = stdout
	138	+	cmd.Stderr = stderr
	139	+	cmd.WaitDelay = cleanup
	140	+	runErr := run(runCtx, cmd)
	141	+	if parent.Err() != nil {
	142	+	return result, parent.Err()
	143	+	}
	144	+	rejected := false
	145	+	for line := range strings.SplitSeq(stdout.buf.String(), "\n") {
	146	+	fields := strings.Split(line, "\t")
	147	+	if len(fields) < 3 \|\| len(fields[0]) != 1 {
	148	+	continue
	149	+	}
	150	+	var status ports.MirrorRefStatus
	151	+	switch fields[0] {
	152	+	case " ", "*":
	153	+	status = ports.MirrorRefOK
	154	+	case "=":
	155	+	status = ports.MirrorRefUpToDate
	156	+	case "!":
	157	+	status = ports.MirrorRefRejected
	158	+	rejected = true
	159	+	default:
	160	+	continue
	161	+	}
	162	+	_, ref, _ := strings.Cut(fields[1], ":")
	163	+	result.Refs = append(result.Refs, ports.MirrorRefOutcome{Ref: ref, Status: status, Reason: safeDetail(fields[2], req.URL)})
	164	+	}
	165	+	if stdout.exceeded \|\| stderr.exceeded {
	166	+	return result, &domain.MirrorError{Class: domain.MirrorClassTransient, Detail: "mirror output limit exceeded"}
	167	+	}
	168	+	if runCtx.Err() != nil {
	169	+	return result, &domain.MirrorError{Class: domain.MirrorClassTransient, Detail: "mirror push timed out"}
	170	+	}
	171	+	if runErr != nil \|\| rejected {
	172	+	class := domain.ClassifyMirrorOutput(stderr.buf.String()+"\n"+stdout.buf.String(), runErr, false)
	173	+	if rejected && class == domain.MirrorClassNone {
	174	+	class = domain.MirrorClassDiverged
	175	+	}
	176	+	var execErr *exec.Error
	177	+	if errors.As(runErr, &execErr) \|\| localGitError(stderr.buf.String()) {
	178	+	class = domain.MirrorClassInternal
	179	+	}
	180	+	detail := safeDetail(stderr.buf.String(), req.URL)
	181	+	if detail == "" {
	182	+	detail = "git push failed"
	183	+	}
	184	+	return result, &domain.MirrorError{Class: class, Detail: detail}
	185	+	}
	186	+	return result, nil
	187	+	}
	188	+
	189	+	func localGitError(stderr string) bool {
	190	+	for _, marker := range []string{"cannot change to", "not a git repository", "src refspec", "bad object", "unable to read"} {
	191	+	if strings.Contains(strings.ToLower(stderr), marker) {
	192	+	return true
	193	+	}
	194	+	}
	195	+	return false
	196	+	}
	197	+
	198	+	// safeDetail removes the complete target before taking a bounded stderr tail.
	199	+	// SSH diagnostics may repeat the user separately, so suppress user@ tokens too.
	200	+	func safeDetail(s, url string) string {
	201	+	s = strings.ReplaceAll(s, url, "[mirror target]")
	202	+	words := strings.Fields(s)
	203	+	for i, w := range words {
	204	+	if strings.Contains(w, "@") \|\| strings.Contains(w, "://") {
	205	+	words[i] = "[redacted]"
	206	+	}
	207	+	}
	208	+	s = strings.Join(words, " ")
	209	+	if len(s) > 4096 {
	210	+	s = s[len(s)-4096:]
	211	+	}
	212	+	return s
	213	+	}
	214	+
	215	+	type limitWriter struct {
	216	+	buf bytes.Buffer
	217	+	max int
	218	+	exceeded bool
	219	+	cancel context.CancelFunc
	220	+	}
	221	+
	222	+	func (w *limitWriter) Write(p []byte) (int, error) {
	223	+	if len(p) > w.max-w.buf.Len() {
	224	+	w.exceeded = true
	225	+	w.cancel()
	226	+	return 0, errors.New("output limit reached")
	227	+	}
	228	+	return w.buf.Write(p)
	229	+	}

A internal/adapters/gitpush/gitpush_test.go +159 -0

@@ -0,0 +1,159 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package gitpush
	4	+
	5	+	import (
	6	+	"context"
	7	+	"crypto/ed25519"
	8	+	"crypto/rand"
	9	+	"errors"
	10	+	"os"
	11	+	"os/exec"
	12	+	"path/filepath"
	13	+	"strings"
	14	+	"testing"
	15	+	"time"
	16	+
	17	+	"github.com/stretchr/testify/require"
	18	+
	19	+	"bnema.dev/ermite/internal/domain"
	20	+	"bnema.dev/ermite/internal/ports"
	21	+	)
	22	+
	23	+	func git(t *testing.T, args ...string) string {
	24	+	t.Helper()
	25	+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	26	+	defer cancel()
	27	+	cmd := exec.CommandContext(ctx, "git", args...)
	28	+	cmd.Env = []string{"PATH=" + os.Getenv("PATH"), "HOME=" + t.TempDir(), "GIT_CONFIG_NOSYSTEM=1", "GIT_CONFIG_GLOBAL=/dev/null", "GIT_AUTHOR_NAME=Test", "GIT_AUTHOR_EMAIL=test@example.org", "GIT_COMMITTER_NAME=Test", "GIT_COMMITTER_EMAIL=test@example.org"}
	29	+	out, err := cmd.CombinedOutput()
	30	+	require.NoError(t, err, string(out))
	31	+	return strings.TrimSpace(string(out))
	32	+	}
	33	+	func setup(t *testing.T) (Config, ports.MirrorPush) {
	34	+	t.Helper()
	35	+	root := t.TempDir()
	36	+	source := filepath.Join(root, "source")
	37	+	git(t, "init", "-b", "main", source)
	38	+	git(t, "-C", source, "commit", "--allow-empty", "-m", "first")
	39	+	bin, err := exec.LookPath("git")
	40	+	require.NoError(t, err)
	41	+	_, key, err := ed25519.GenerateKey(rand.Reader)
	42	+	require.NoError(t, err)
	43	+	return Config{Bin: bin, Path: os.Getenv("PATH"), Home: t.TempDir(), KnownHosts: filepath.Join(root, "mirror_known_hosts"), TempDir: filepath.Join(root, "keys"), RepoDir: func(domain.RepoName) string { return source }, AllowedProtocols: "file", Timeout: 5 * time.Second}, ports.MirrorPush{PrivateKey: key, Refspecs: []string{"refs/heads/main:refs/heads/main"}}
	44	+	}
	45	+	func clean(t *testing.T, cfg Config) {
	46	+	t.Helper()
	47	+	entries, err := os.ReadDir(cfg.TempDir)
	48	+	require.NoError(t, err)
	49	+	require.Empty(t, entries)
	50	+	}
	51	+	func TestPushLocal(t *testing.T) {
	52	+	cfg, req := setup(t)
	53	+	remote := filepath.Join(t.TempDir(), "remote.git")
	54	+	git(t, "init", "--bare", remote)
	55	+	req.URL = "file://" + remote
	56	+	p := New(cfg)
	57	+	result, err := p.Push(context.Background(), req)
	58	+	require.NoError(t, err)
	59	+	require.Len(t, result.Refs, 1)
	60	+	require.Equal(t, ports.MirrorRefOK, result.Refs[0].Status)
	61	+	clean(t, cfg)
	62	+	result, err = p.Push(context.Background(), req)
	63	+	require.NoError(t, err)
	64	+	require.Equal(t, ports.MirrorRefUpToDate, result.Refs[0].Status)
	65	+	clean(t, cfg)
	66	+	other := filepath.Join(t.TempDir(), "other")
	67	+	git(t, "init", "-b", "main", other)
	68	+	git(t, "-C", other, "commit", "--allow-empty", "-m", "unrelated")
	69	+	git(t, "-C", other, "push", "file://"+remote, "refs/heads/main:refs/heads/other")
	70	+	// Move remote main locally, without a force push, to establish divergence.
	71	+	otherID := git(t, "-C", other, "rev-parse", "HEAD")
	72	+	git(t, "--git-dir", remote, "update-ref", "refs/heads/main", otherID)
	73	+	_, err = p.Push(context.Background(), req)
	74	+	var me *domain.MirrorError
	75	+	require.ErrorAs(t, err, &me)
	76	+	require.Equal(t, domain.MirrorClassDiverged, me.Class)
	77	+	require.Equal(t, otherID, git(t, "--git-dir", remote, "rev-parse", "refs/heads/main"))
	78	+	clean(t, cfg)
	79	+	req.URL = "file://" + filepath.Join(t.TempDir(), "missing.git")
	80	+	_, err = p.Push(context.Background(), req)
	81	+	require.ErrorAs(t, err, &me)
	82	+	require.Equal(t, domain.MirrorClassRemoteNotFound, me.Class)
	83	+	clean(t, cfg)
	84	+	}
	85	+	func TestBoundsAndKeyEnvironment(t *testing.T) {
	86	+	for _, mode := range []string{"success", "error", "timeout", "stdout", "stderr", "cancel"} {
	87	+	t.Run(mode, func(t *testing.T) {
	88	+	cfg, req := setup(t)
	89	+	req.URL = "file:///unused"
	90	+	script := filepath.Join(t.TempDir(), "git")
	91	+	report := filepath.Join(t.TempDir(), "env")
	92	+	body := "#!/bin/sh\nenv > '" + report + "'\nfor d in '" + cfg.TempDir + "'/push-; do\n test \"$(stat -c %a \"$d\")\" = 700 \|\| exit 90\n for k in \"$d\"/key-; do test \"$(stat -c %a \"$k\")\" = 600 \|\| exit 91; done\ndone\n"
	93	+	switch mode {
	94	+	case "error":
	95	+	body += "echo 'Permission denied (publickey)' >&2; exit 1\n"
	96	+	case "timeout", "cancel":
	97	+	body += "sleep 10\n"
	98	+	case "stdout":
	99	+	body += "head -c 4096 /dev/zero\n"
	100	+	case "stderr":
	101	+	body += "head -c 4096 /dev/zero >&2\n"
	102	+	}
	103	+	require.NoError(t, os.WriteFile(script, []byte(body), 0700))
	104	+	cfg.Bin = script
	105	+	cfg.MaxOutput = 1024
	106	+	cfg.Timeout = 200 * time.Millisecond
	107	+	t.Setenv("ERMITE_PRIVATE_TEST", "not-forwarded")
	108	+	ctx := context.Background()
	109	+	if mode == "cancel" {
	110	+	var cancel context.CancelFunc
	111	+	ctx, cancel = context.WithTimeout(ctx, 50*time.Millisecond)
	112	+	defer cancel()
	113	+	}
	114	+	_, err := New(cfg).Push(ctx, req)
	115	+	switch mode {
	116	+	case "success":
	117	+	require.NoError(t, err)
	118	+	case "cancel":
	119	+	require.ErrorIs(t, err, context.DeadlineExceeded)
	120	+	default:
	121	+	var me *domain.MirrorError
	122	+	require.ErrorAs(t, err, &me)
	123	+	want := domain.MirrorClassTransient
	124	+	if mode == "error" {
	125	+	want = domain.MirrorClassAuthDenied
	126	+	}
	127	+	require.Equal(t, want, me.Class)
	128	+	}
	129	+	clean(t, cfg)
	130	+	env, e := os.ReadFile(report)
	131	+	require.NoError(t, e)
	132	+	require.NotContains(t, string(env), "ERMITE_")
	133	+	require.Contains(t, string(env), "IdentitiesOnly=yes")
	134	+	require.Contains(t, string(env), "BatchMode=yes")
	135	+	require.Contains(t, string(env), "StrictHostKeyChecking=accept-new")
	136	+	require.Contains(t, string(env), "UserKnownHostsFile=")
	137	+	})
	138	+	}
	139	+	}
	140	+	func TestClassifierSamples(t *testing.T) {
	141	+	for _, tc := range []struct {
	142	+	s string
	143	+	c domain.MirrorFailureClass
	144	+	}{{"Permission denied (publickey)", domain.MirrorClassAuthDenied}, {"Repository not found", domain.MirrorClassRemoteNotFound}, {"Host key verification failed", domain.MirrorClassHostKey}, {"Could not resolve hostname", domain.MirrorClassTransient}, {"non-fast-forward", domain.MirrorClassDiverged}} {
	145	+	require.Equal(t, tc.c, domain.ClassifyMirrorOutput(tc.s, errors.New("exit 1"), false))
	146	+	}
	147	+	}
	148	+	func TestRejectUnsafeRefspecs(t *testing.T) {
	149	+	for _, ref := range []string{"+refs/heads/main:refs/heads/main", ":refs/heads/main", "--mirror", "refs/heads/:refs/heads/", "refs/heads/main:refs/heads/other"} {
	150	+	require.False(t, validRefspec(ref))
	151	+	}
	152	+	require.True(t, validRefspec("refs/tags/v1:refs/tags/v1"))
	153	+	}
	154	+	func TestSafeDetail(t *testing.T) {
	155	+	s := safeDetail(strings.Repeat("x", 5000)+" ssh://user:password@host/repo user@host", "ssh://user:password@host/repo")
	156	+	require.LessOrEqual(t, len(s), 4096)
	157	+	require.NotContains(t, s, "password")
	158	+	require.NotContains(t, s, "user@")
	159	+	}

@@ -0,0 +1,159 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package gitpush
	4	+
	5	+	import (
	6	+	"context"
	7	+	"crypto/ed25519"
	8	+	"crypto/rand"
	9	+	"errors"
	10	+	"os"
	11	+	"os/exec"
	12	+	"path/filepath"
	13	+	"strings"
	14	+	"testing"
	15	+	"time"
	16	+
	17	+	"github.com/stretchr/testify/require"
	18	+
	19	+	"bnema.dev/ermite/internal/domain"
	20	+	"bnema.dev/ermite/internal/ports"
	21	+	)
	22	+
	23	+	func git(t *testing.T, args ...string) string {
	24	+	t.Helper()
	25	+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	26	+	defer cancel()
	27	+	cmd := exec.CommandContext(ctx, "git", args...)
	28	+	cmd.Env = []string{"PATH=" + os.Getenv("PATH"), "HOME=" + t.TempDir(), "GIT_CONFIG_NOSYSTEM=1", "GIT_CONFIG_GLOBAL=/dev/null", "GIT_AUTHOR_NAME=Test", "GIT_AUTHOR_EMAIL=test@example.org", "GIT_COMMITTER_NAME=Test", "GIT_COMMITTER_EMAIL=test@example.org"}
	29	+	out, err := cmd.CombinedOutput()
	30	+	require.NoError(t, err, string(out))
	31	+	return strings.TrimSpace(string(out))
	32	+	}
	33	+	func setup(t *testing.T) (Config, ports.MirrorPush) {
	34	+	t.Helper()
	35	+	root := t.TempDir()
	36	+	source := filepath.Join(root, "source")
	37	+	git(t, "init", "-b", "main", source)
	38	+	git(t, "-C", source, "commit", "--allow-empty", "-m", "first")
	39	+	bin, err := exec.LookPath("git")
	40	+	require.NoError(t, err)
	41	+	_, key, err := ed25519.GenerateKey(rand.Reader)
	42	+	require.NoError(t, err)
	43	+	return Config{Bin: bin, Path: os.Getenv("PATH"), Home: t.TempDir(), KnownHosts: filepath.Join(root, "mirror_known_hosts"), TempDir: filepath.Join(root, "keys"), RepoDir: func(domain.RepoName) string { return source }, AllowedProtocols: "file", Timeout: 5 * time.Second}, ports.MirrorPush{PrivateKey: key, Refspecs: []string{"refs/heads/main:refs/heads/main"}}
	44	+	}
	45	+	func clean(t *testing.T, cfg Config) {
	46	+	t.Helper()
	47	+	entries, err := os.ReadDir(cfg.TempDir)
	48	+	require.NoError(t, err)
	49	+	require.Empty(t, entries)
	50	+	}
	51	+	func TestPushLocal(t *testing.T) {
	52	+	cfg, req := setup(t)
	53	+	remote := filepath.Join(t.TempDir(), "remote.git")
	54	+	git(t, "init", "--bare", remote)
	55	+	req.URL = "file://" + remote
	56	+	p := New(cfg)
	57	+	result, err := p.Push(context.Background(), req)
	58	+	require.NoError(t, err)
	59	+	require.Len(t, result.Refs, 1)
	60	+	require.Equal(t, ports.MirrorRefOK, result.Refs[0].Status)
	61	+	clean(t, cfg)
	62	+	result, err = p.Push(context.Background(), req)
	63	+	require.NoError(t, err)
	64	+	require.Equal(t, ports.MirrorRefUpToDate, result.Refs[0].Status)
	65	+	clean(t, cfg)
	66	+	other := filepath.Join(t.TempDir(), "other")
	67	+	git(t, "init", "-b", "main", other)
	68	+	git(t, "-C", other, "commit", "--allow-empty", "-m", "unrelated")
	69	+	git(t, "-C", other, "push", "file://"+remote, "refs/heads/main:refs/heads/other")
	70	+	// Move remote main locally, without a force push, to establish divergence.
	71	+	otherID := git(t, "-C", other, "rev-parse", "HEAD")
	72	+	git(t, "--git-dir", remote, "update-ref", "refs/heads/main", otherID)
	73	+	_, err = p.Push(context.Background(), req)
	74	+	var me *domain.MirrorError
	75	+	require.ErrorAs(t, err, &me)
	76	+	require.Equal(t, domain.MirrorClassDiverged, me.Class)
	77	+	require.Equal(t, otherID, git(t, "--git-dir", remote, "rev-parse", "refs/heads/main"))
	78	+	clean(t, cfg)
	79	+	req.URL = "file://" + filepath.Join(t.TempDir(), "missing.git")
	80	+	_, err = p.Push(context.Background(), req)
	81	+	require.ErrorAs(t, err, &me)
	82	+	require.Equal(t, domain.MirrorClassRemoteNotFound, me.Class)
	83	+	clean(t, cfg)
	84	+	}
	85	+	func TestBoundsAndKeyEnvironment(t *testing.T) {
	86	+	for _, mode := range []string{"success", "error", "timeout", "stdout", "stderr", "cancel"} {
	87	+	t.Run(mode, func(t *testing.T) {
	88	+	cfg, req := setup(t)
	89	+	req.URL = "file:///unused"
	90	+	script := filepath.Join(t.TempDir(), "git")
	91	+	report := filepath.Join(t.TempDir(), "env")
	92	+	body := "#!/bin/sh\nenv > '" + report + "'\nfor d in '" + cfg.TempDir + "'/push-; do\n test \"$(stat -c %a \"$d\")\" = 700 \|\| exit 90\n for k in \"$d\"/key-; do test \"$(stat -c %a \"$k\")\" = 600 \|\| exit 91; done\ndone\n"
	93	+	switch mode {
	94	+	case "error":
	95	+	body += "echo 'Permission denied (publickey)' >&2; exit 1\n"
	96	+	case "timeout", "cancel":
	97	+	body += "sleep 10\n"
	98	+	case "stdout":
	99	+	body += "head -c 4096 /dev/zero\n"
	100	+	case "stderr":
	101	+	body += "head -c 4096 /dev/zero >&2\n"
	102	+	}
	103	+	require.NoError(t, os.WriteFile(script, []byte(body), 0700))
	104	+	cfg.Bin = script
	105	+	cfg.MaxOutput = 1024
	106	+	cfg.Timeout = 200 * time.Millisecond
	107	+	t.Setenv("ERMITE_PRIVATE_TEST", "not-forwarded")
	108	+	ctx := context.Background()
	109	+	if mode == "cancel" {
	110	+	var cancel context.CancelFunc
	111	+	ctx, cancel = context.WithTimeout(ctx, 50*time.Millisecond)
	112	+	defer cancel()
	113	+	}
	114	+	_, err := New(cfg).Push(ctx, req)
	115	+	switch mode {
	116	+	case "success":
	117	+	require.NoError(t, err)
	118	+	case "cancel":
	119	+	require.ErrorIs(t, err, context.DeadlineExceeded)
	120	+	default:
	121	+	var me *domain.MirrorError
	122	+	require.ErrorAs(t, err, &me)
	123	+	want := domain.MirrorClassTransient
	124	+	if mode == "error" {
	125	+	want = domain.MirrorClassAuthDenied
	126	+	}
	127	+	require.Equal(t, want, me.Class)
	128	+	}
	129	+	clean(t, cfg)
	130	+	env, e := os.ReadFile(report)
	131	+	require.NoError(t, e)
	132	+	require.NotContains(t, string(env), "ERMITE_")
	133	+	require.Contains(t, string(env), "IdentitiesOnly=yes")
	134	+	require.Contains(t, string(env), "BatchMode=yes")
	135	+	require.Contains(t, string(env), "StrictHostKeyChecking=accept-new")
	136	+	require.Contains(t, string(env), "UserKnownHostsFile=")
	137	+	})
	138	+	}
	139	+	}
	140	+	func TestClassifierSamples(t *testing.T) {
	141	+	for _, tc := range []struct {
	142	+	s string
	143	+	c domain.MirrorFailureClass
	144	+	}{{"Permission denied (publickey)", domain.MirrorClassAuthDenied}, {"Repository not found", domain.MirrorClassRemoteNotFound}, {"Host key verification failed", domain.MirrorClassHostKey}, {"Could not resolve hostname", domain.MirrorClassTransient}, {"non-fast-forward", domain.MirrorClassDiverged}} {
	145	+	require.Equal(t, tc.c, domain.ClassifyMirrorOutput(tc.s, errors.New("exit 1"), false))
	146	+	}
	147	+	}
	148	+	func TestRejectUnsafeRefspecs(t *testing.T) {
	149	+	for _, ref := range []string{"+refs/heads/main:refs/heads/main", ":refs/heads/main", "--mirror", "refs/heads/:refs/heads/", "refs/heads/main:refs/heads/other"} {
	150	+	require.False(t, validRefspec(ref))
	151	+	}
	152	+	require.True(t, validRefspec("refs/tags/v1:refs/tags/v1"))
	153	+	}
	154	+	func TestSafeDetail(t *testing.T) {
	155	+	s := safeDetail(strings.Repeat("x", 5000)+" ssh://user:password@host/repo user@host", "ssh://user:password@host/repo")
	156	+	require.LessOrEqual(t, len(s), 4096)
	157	+	require.NotContains(t, s, "password")
	158	+	require.NotContains(t, s, "user@")
	159	+	}

A internal/adapters/gitpush/run_linux.go +107 -0

@@ -0,0 +1,107 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	//go:build linux
	4	+
	5	+	package gitpush
	6	+
	7	+	import (
	8	+	"context"
	9	+	"errors"
	10	+	"fmt"
	11	+	"os/exec"
	12	+	"syscall"
	13	+
	14	+	"golang.org/x/sys/unix"
	15	+	)
	16	+
	17	+	// Tests replace these two.
	18	+	var (
	19	+	waitid = unix.Waitid
	20	+	killGroup = func(pgid int) error { return syscall.Kill(-pgid, syscall.SIGKILL) }
	21	+	)
	22	+
	23	+	// run starts cmd, waits for it, and kills what it leaves running.
	24	+	//
	25	+	// git runs git-remote-https, which is the process that connects and that holds
	26	+	// the pipes of git's output. Killing git alone leaves it running, and Wait
	27	+	// blocked on the pipes. So git leads a process group of its own, and the group
	28	+	// is killed when ctx is done and also when git has exited.
	29	+	//
	30	+	// The invariant that makes killing a group safe: the group is only signalled
	31	+	// while its leader is still not reaped. Linux does not reuse a PID as long as
	32	+	// the process is a zombie nobody waited for, nor while it is the group ID of a
	33	+	// live member, so -pid is still our group and cannot be another one. The
	34	+	// leader is therefore waited for with waitid(WNOWAIT), which observes its exit
	35	+	// without reaping it; the group is signalled; and only then cmd.Wait reaps. A
	36	+	// kill(-pid) after Wait, or from Cmd.Cancel which races with it, would not be
	37	+	// safe. When waitid fails, whatever the reason, nothing proves the leader is
	38	+	// unreaped: the group is not signalled, only the leader is killed (through
	39	+	// its pidfd, which cannot hit a reused PID), and the error is returned.
	40	+	// cmd.WaitDelay (set by the caller) stays the backstop for a descendant that
	41	+	// escaped the group and holds the pipes.
	42	+	func run(ctx context.Context, cmd *exec.Cmd) error {
	43	+	if err := ctx.Err(); err != nil {
	44	+	return err
	45	+	}
	46	+	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
	47	+	if err := cmd.Start(); err != nil {
	48	+	return err
	49	+	}
	50	+	pid := cmd.Process.Pid
	51	+	observed := make(chan error, 1)
	52	+	go func() { observed <- waitExit(pid) }()
	53	+
	54	+	var werr error
	55	+	select {
	56	+	case err := <-observed:
	57	+	if err != nil {
	58	+	werr = err // the leader may have been reaped: no group signal
	59	+	} else {
	60	+	signalGroup(cmd, pid) // exited, not reaped: a zombie
	61	+	}
	62	+	case <-ctx.Done():
	63	+	select {
	64	+	case err := <-observed:
	65	+	// waitid answered at the same time: trust it as above.
	66	+	if err != nil {
	67	+	werr = err
	68	+	} else {
	69	+	signalGroup(cmd, pid)
	70	+	}
	71	+	default:
	72	+	// No waitid error so far: the leader is alive or a zombie, unreaped
	73	+	// (only cmd.Wait below reaps it).
	74	+	signalGroup(cmd, pid)
	75	+	if err := <-observed; err != nil {
	76	+	werr = err // after the signals: reported, nothing more to do
	77	+	}
	78	+	}
	79	+	}
	80	+	if werr != nil {
	81	+	_ = cmd.Process.Kill()
	82	+	}
	83	+	err := cmd.Wait()
	84	+	if werr != nil {
	85	+	return fmt.Errorf("waiting for the process %d: %w", pid, werr)
	86	+	}
	87	+	return err
	88	+	}
	89	+
	90	+	// signalGroup kills the group of the leader, and the leader alone should it
	91	+	// have left the group. The leader must not be reaped yet.
	92	+	func signalGroup(cmd *exec.Cmd, pid int) {
	93	+	_ = killGroup(pid)
	94	+	_ = cmd.Process.Kill()
	95	+	}
	96	+
	97	+	// waitExit returns when the child pid has exited, without reaping it, or the
	98	+	// error of waitid.
	99	+	func waitExit(pid int) error {
	100	+	var info unix.Siginfo
	101	+	for {
	102	+	err := waitid(unix.P_PID, pid, &info, unix.WEXITED\|unix.WNOWAIT, nil)
	103	+	if !errors.Is(err, unix.EINTR) {
	104	+	return err
	105	+	}
	106	+	}
	107	+	}

@@ -0,0 +1,107 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	//go:build linux
	4	+
	5	+	package gitpush
	6	+
	7	+	import (
	8	+	"context"
	9	+	"errors"
	10	+	"fmt"
	11	+	"os/exec"
	12	+	"syscall"
	13	+
	14	+	"golang.org/x/sys/unix"
	15	+	)
	16	+
	17	+	// Tests replace these two.
	18	+	var (
	19	+	waitid = unix.Waitid
	20	+	killGroup = func(pgid int) error { return syscall.Kill(-pgid, syscall.SIGKILL) }
	21	+	)
	22	+
	23	+	// run starts cmd, waits for it, and kills what it leaves running.
	24	+	//
	25	+	// git runs git-remote-https, which is the process that connects and that holds
	26	+	// the pipes of git's output. Killing git alone leaves it running, and Wait
	27	+	// blocked on the pipes. So git leads a process group of its own, and the group
	28	+	// is killed when ctx is done and also when git has exited.
	29	+	//
	30	+	// The invariant that makes killing a group safe: the group is only signalled
	31	+	// while its leader is still not reaped. Linux does not reuse a PID as long as
	32	+	// the process is a zombie nobody waited for, nor while it is the group ID of a
	33	+	// live member, so -pid is still our group and cannot be another one. The
	34	+	// leader is therefore waited for with waitid(WNOWAIT), which observes its exit
	35	+	// without reaping it; the group is signalled; and only then cmd.Wait reaps. A
	36	+	// kill(-pid) after Wait, or from Cmd.Cancel which races with it, would not be
	37	+	// safe. When waitid fails, whatever the reason, nothing proves the leader is
	38	+	// unreaped: the group is not signalled, only the leader is killed (through
	39	+	// its pidfd, which cannot hit a reused PID), and the error is returned.
	40	+	// cmd.WaitDelay (set by the caller) stays the backstop for a descendant that
	41	+	// escaped the group and holds the pipes.
	42	+	func run(ctx context.Context, cmd *exec.Cmd) error {
	43	+	if err := ctx.Err(); err != nil {
	44	+	return err
	45	+	}
	46	+	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
	47	+	if err := cmd.Start(); err != nil {
	48	+	return err
	49	+	}
	50	+	pid := cmd.Process.Pid
	51	+	observed := make(chan error, 1)
	52	+	go func() { observed <- waitExit(pid) }()
	53	+
	54	+	var werr error
	55	+	select {
	56	+	case err := <-observed:
	57	+	if err != nil {
	58	+	werr = err // the leader may have been reaped: no group signal
	59	+	} else {
	60	+	signalGroup(cmd, pid) // exited, not reaped: a zombie
	61	+	}
	62	+	case <-ctx.Done():
	63	+	select {
	64	+	case err := <-observed:
	65	+	// waitid answered at the same time: trust it as above.
	66	+	if err != nil {
	67	+	werr = err
	68	+	} else {
	69	+	signalGroup(cmd, pid)
	70	+	}
	71	+	default:
	72	+	// No waitid error so far: the leader is alive or a zombie, unreaped
	73	+	// (only cmd.Wait below reaps it).
	74	+	signalGroup(cmd, pid)
	75	+	if err := <-observed; err != nil {
	76	+	werr = err // after the signals: reported, nothing more to do
	77	+	}
	78	+	}
	79	+	}
	80	+	if werr != nil {
	81	+	_ = cmd.Process.Kill()
	82	+	}
	83	+	err := cmd.Wait()
	84	+	if werr != nil {
	85	+	return fmt.Errorf("waiting for the process %d: %w", pid, werr)
	86	+	}
	87	+	return err
	88	+	}
	89	+
	90	+	// signalGroup kills the group of the leader, and the leader alone should it
	91	+	// have left the group. The leader must not be reaped yet.
	92	+	func signalGroup(cmd *exec.Cmd, pid int) {
	93	+	_ = killGroup(pid)
	94	+	_ = cmd.Process.Kill()
	95	+	}
	96	+
	97	+	// waitExit returns when the child pid has exited, without reaping it, or the
	98	+	// error of waitid.
	99	+	func waitExit(pid int) error {
	100	+	var info unix.Siginfo
	101	+	for {
	102	+	err := waitid(unix.P_PID, pid, &info, unix.WEXITED\|unix.WNOWAIT, nil)
	103	+	if !errors.Is(err, unix.EINTR) {
	104	+	return err
	105	+	}
	106	+	}
	107	+	}

A internal/adapters/kvstore/mirror.go +190 -0

@@ -0,0 +1,190 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package kvstore
	4	+
	5	+	import (
	6	+	"context"
	7	+	"errors"
	8	+	"fmt"
	9	+	"strings"
	10	+	"time"
	11	+
	12	+	kv "bnema.dev/ermite-kv"
	13	+	"google.golang.org/protobuf/proto"
	14	+
	15	+	storagepb "bnema.dev/ermite/gen/ermite/storage"
	16	+	"bnema.dev/ermite/internal/domain"
	17	+	"bnema.dev/ermite/internal/ports"
	18	+	)
	19	+
	20	+	var _ ports.MirrorStore = (*Store)(nil)
	21	+	var _ ports.MirrorKeyStore = (*Store)(nil)
	22	+
	23	+	const mirrorConfigKey = "mirror/config"
	24	+	const mirrorRepoPrefix = "mirror/repo/"
	25	+	const mirrorStatePrefix = "mirror/state/"
	26	+	const mirrorSnapshotPrefix = "mirror/snapshot/"
	27	+
	28	+	func mirrorKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName) [][]byte {
	29	+	var keys [][]byte
	30	+	for _, prefix := range []string{mirrorRepoPrefix, mirrorStatePrefix, mirrorSnapshotPrefix} {
	31	+	key := []byte(prefix + repo.String())
	32	+	if _, ok := snap.Get(ctx, key); ok {
	33	+	keys = append(keys, key)
	34	+	}
	35	+	}
	36	+	return keys
	37	+	}
	38	+
	39	+	func (s *Store) readMirror(ctx context.Context, key string, pb proto.Message) error {
	40	+	item, ok := s.db.Get(ctx, []byte(key))
	41	+	if !ok {
	42	+	return domain.ErrNotFound
	43	+	}
	44	+	if err := proto.Unmarshal(item.Value, pb); err != nil {
	45	+	return fmt.Errorf("decode mirror record: %w", err)
	46	+	}
	47	+	return nil
	48	+	}
	49	+
	50	+	func (s *Store) writeMirror(ctx context.Context, key string, repo domain.RepoName, pb proto.Message) error {
	51	+	val := mustMarshal(pb)
	52	+	for range updateAttempts {
	53	+	var b kv.Batch
	54	+	if key != mirrorConfigKey {
	55	+	if err := guardRepoKey(ctx, s.db, &b, repo); err != nil {
	56	+	return err
	57	+	}
	58	+	}
	59	+	b.Put([]byte(key), val)
	60	+	_, err := s.db.Apply(ctx, &b)
	61	+	if errors.Is(err, kv.ErrConflict) {
	62	+	continue
	63	+	}
	64	+	if err != nil {
	65	+	return fmt.Errorf("store mirror record: %w", err)
	66	+	}
	67	+	return nil
	68	+	}
	69	+	return domain.ErrConflict
	70	+	}
	71	+
	72	+	func (s *Store) GetGlobal(ctx context.Context) (domain.MirrorGlobalConfig, error) {
	73	+	var pb storagepb.MirrorGlobal
	74	+	err := s.readMirror(ctx, mirrorConfigKey, &pb)
	75	+	if errors.Is(err, domain.ErrNotFound) {
	76	+	err = nil
	77	+	}
	78	+	return domain.MirrorGlobalConfig{Enabled: pb.Enabled, URLTemplate: pb.UrlTemplate, Cron: pb.Cron, AllBranches: pb.AllBranches, UpdatedAt: decodeTime(pb.UpdatedAt)}, err
	79	+	}
	80	+	func (s *Store) PutGlobal(ctx context.Context, g domain.MirrorGlobalConfig) error {
	81	+	return s.writeMirror(ctx, mirrorConfigKey, domain.RepoName{}, &storagepb.MirrorGlobal{Enabled: g.Enabled, UrlTemplate: g.URLTemplate, Cron: g.Cron, AllBranches: g.AllBranches, UpdatedAt: encodeTime(g.UpdatedAt)})
	82	+	}
	83	+	func decodeMirrorRepo(repo domain.RepoName, pb *storagepb.MirrorRepo) domain.MirrorRepoConfig {
	84	+	return domain.MirrorRepoConfig{Repo: repo, URL: pb.Url, Cron: pb.Cron, AllBranches: pb.AllBranches, Excluded: pb.Excluded, UpdatedAt: decodeTime(pb.UpdatedAt)}
	85	+	}
	86	+	func (s *Store) GetRepoConfig(ctx context.Context, repo domain.RepoName) (domain.MirrorRepoConfig, error) {
	87	+	var pb storagepb.MirrorRepo
	88	+	err := s.readMirror(ctx, mirrorRepoPrefix+repo.String(), &pb)
	89	+	return decodeMirrorRepo(repo, &pb), err
	90	+	}
	91	+	func (s *Store) PutRepoConfig(ctx context.Context, c domain.MirrorRepoConfig) error {
	92	+	return s.writeMirror(ctx, mirrorRepoPrefix+c.Repo.String(), c.Repo, &storagepb.MirrorRepo{Url: c.URL, Cron: c.Cron, AllBranches: c.AllBranches, Excluded: c.Excluded, UpdatedAt: encodeTime(c.UpdatedAt)})
	93	+	}
	94	+	func (s *Store) DeleteRepoConfig(ctx context.Context, repo domain.RepoName) error {
	95	+	var b kv.Batch
	96	+	b.Delete([]byte(mirrorRepoPrefix + repo.String()))
	97	+	_, err := s.db.Apply(ctx, &b)
	98	+	if err != nil {
	99	+	return fmt.Errorf("delete mirror config: %w", err)
	100	+	}
	101	+	return nil
	102	+	}
	103	+	func (s *Store) ListRepoConfigs(ctx context.Context) ([]domain.MirrorRepoConfig, error) {
	104	+	var out []domain.MirrorRepoConfig
	105	+	for key, item := range s.db.Snapshot(ctx).Prefix(ctx, []byte(mirrorRepoPrefix)) {
	106	+	repo, err := domain.ParseRepoName(strings.TrimPrefix(string(key), mirrorRepoPrefix))
	107	+	if err != nil {
	108	+	return nil, err
	109	+	}
	110	+	var pb storagepb.MirrorRepo
	111	+	if err := proto.Unmarshal(item.Value, &pb); err != nil {
	112	+	return nil, fmt.Errorf("decode mirror config: %w", err)
	113	+	}
	114	+	out = append(out, decodeMirrorRepo(repo, &pb))
	115	+	}
	116	+	return out, nil
	117	+	}
	118	+	func decodeMirrorStatus(repo domain.RepoName, pb *storagepb.MirrorState) domain.MirrorStatus {
	119	+	return domain.MirrorStatus{Repo: repo, State: domain.MirrorState(pb.State), Class: domain.MirrorFailureClass(pb.Class), Reason: pb.Reason, Detail: pb.Detail, ConsecutiveFailures: int(pb.ConsecutiveFailures), LastAttemptAt: decodeTime(pb.LastAttemptAt), LastSuccessAt: decodeTime(pb.LastSuccessAt), NextRetryAt: decodeTime(pb.NextRetryAt), PausedAt: decodeTime(pb.PausedAt), UpdatedAt: decodeTime(pb.UpdatedAt)}
	120	+	}
	121	+	func (s *Store) GetStatus(ctx context.Context, repo domain.RepoName) (domain.MirrorStatus, error) {
	122	+	var pb storagepb.MirrorState
	123	+	err := s.readMirror(ctx, mirrorStatePrefix+repo.String(), &pb)
	124	+	return decodeMirrorStatus(repo, &pb), err
	125	+	}
	126	+	func (s *Store) PutStatus(ctx context.Context, c domain.MirrorStatus) error {
	127	+	return s.writeMirror(ctx, mirrorStatePrefix+c.Repo.String(), c.Repo, &storagepb.MirrorState{State: string(c.State), Class: string(c.Class), Reason: c.Reason, Detail: c.Detail, ConsecutiveFailures: int32(c.ConsecutiveFailures), LastAttemptAt: encodeTime(c.LastAttemptAt), LastSuccessAt: encodeTime(c.LastSuccessAt), NextRetryAt: encodeTime(c.NextRetryAt), PausedAt: encodeTime(c.PausedAt), UpdatedAt: encodeTime(c.UpdatedAt)})
	128	+	}
	129	+	func (s *Store) ListStatus(ctx context.Context) ([]domain.MirrorStatus, error) {
	130	+	var out []domain.MirrorStatus
	131	+	for key, item := range s.db.Snapshot(ctx).Prefix(ctx, []byte(mirrorStatePrefix)) {
	132	+	repo, err := domain.ParseRepoName(strings.TrimPrefix(string(key), mirrorStatePrefix))
	133	+	if err != nil {
	134	+	return nil, err
	135	+	}
	136	+	var pb storagepb.MirrorState
	137	+	if err := proto.Unmarshal(item.Value, &pb); err != nil {
	138	+	return nil, fmt.Errorf("decode mirror state: %w", err)
	139	+	}
	140	+	out = append(out, decodeMirrorStatus(repo, &pb))
	141	+	}
	142	+	return out, nil
	143	+	}
	144	+	func (s *Store) GetSnapshot(ctx context.Context, repo domain.RepoName) (domain.MirrorSnapshot, error) {
	145	+	var pb storagepb.MirrorSnapshot
	146	+	err := s.readMirror(ctx, mirrorSnapshotPrefix+repo.String(), &pb)
	147	+	out := domain.MirrorSnapshot{URL: pb.Url, AllBranches: pb.AllBranches, At: decodeTime(pb.At)}
	148	+	for _, ref := range pb.Refs {
	149	+	id, parseErr := domain.ParseObjectID(ref.Id)
	150	+	if parseErr != nil {
	151	+	return domain.MirrorSnapshot{}, fmt.Errorf("decode mirror snapshot ref: %w", parseErr)
	152	+	}
	153	+	out.Refs = append(out.Refs, domain.MirrorRef{Name: ref.Name, ID: id})
	154	+	}
	155	+	return out, err
	156	+	}
	157	+	func (s *Store) PutSnapshot(ctx context.Context, repo domain.RepoName, c domain.MirrorSnapshot) error {
	158	+	pb := &storagepb.MirrorSnapshot{Url: c.URL, AllBranches: c.AllBranches, At: encodeTime(c.At)}
	159	+	for _, ref := range c.Refs {
	160	+	pb.Refs = append(pb.Refs, &storagepb.MirrorSnapshotRef{Name: ref.Name, Id: ref.ID.String()})
	161	+	}
	162	+	return s.writeMirror(ctx, mirrorSnapshotPrefix+repo.String(), repo, pb)
	163	+	}
	164	+	func (s *Store) GetMirrorKey(ctx context.Context) ([]byte, time.Time, error) {
	165	+	var pb storagepb.MirrorKey
	166	+	err := s.readMirror(ctx, domain.MirrorKeyPath, &pb)
	167	+	return pb.Sealed, decodeTime(pb.CreatedAt), err
	168	+	}
	169	+	func (s *Store) PutMirrorKey(ctx context.Context, sealed []byte, at time.Time, ifAbsent bool) error {
	170	+	var b kv.Batch
	171	+	key := []byte(domain.MirrorKeyPath)
	172	+	if ifAbsent {
	173	+	b.RequireAbsent(key)
	174	+	}
	175	+	b.Put(key, mustMarshal(&storagepb.MirrorKey{Sealed: sealed, CreatedAt: encodeTime(at)}))
	176	+	_, err := s.db.Apply(ctx, &b)
	177	+	if ifAbsent && errors.Is(err, kv.ErrConflict) {
	178	+	return domain.ErrAlreadyExists
	179	+	}
	180	+	if err != nil {
	181	+	return fmt.Errorf("store mirror key: %w", err)
	182	+	}
	183	+	return nil
	184	+	}
	185	+
	186	+	// RawMirrorKey exposes the stored protobuf for encryption-at-rest tests.
	187	+	func (s *Store) RawMirrorKey(ctx context.Context) ([]byte, bool) {
	188	+	item, ok := s.db.Get(ctx, []byte(domain.MirrorKeyPath))
	189	+	return item.Value, ok
	190	+	}

@@ -0,0 +1,190 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package kvstore
	4	+
	5	+	import (
	6	+	"context"
	7	+	"errors"
	8	+	"fmt"
	9	+	"strings"
	10	+	"time"
	11	+
	12	+	kv "bnema.dev/ermite-kv"
	13	+	"google.golang.org/protobuf/proto"
	14	+
	15	+	storagepb "bnema.dev/ermite/gen/ermite/storage"
	16	+	"bnema.dev/ermite/internal/domain"
	17	+	"bnema.dev/ermite/internal/ports"
	18	+	)
	19	+
	20	+	var _ ports.MirrorStore = (*Store)(nil)
	21	+	var _ ports.MirrorKeyStore = (*Store)(nil)
	22	+
	23	+	const mirrorConfigKey = "mirror/config"
	24	+	const mirrorRepoPrefix = "mirror/repo/"
	25	+	const mirrorStatePrefix = "mirror/state/"
	26	+	const mirrorSnapshotPrefix = "mirror/snapshot/"
	27	+
	28	+	func mirrorKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName) [][]byte {
	29	+	var keys [][]byte
	30	+	for _, prefix := range []string{mirrorRepoPrefix, mirrorStatePrefix, mirrorSnapshotPrefix} {
	31	+	key := []byte(prefix + repo.String())
	32	+	if _, ok := snap.Get(ctx, key); ok {
	33	+	keys = append(keys, key)
	34	+	}
	35	+	}
	36	+	return keys
	37	+	}
	38	+
	39	+	func (s *Store) readMirror(ctx context.Context, key string, pb proto.Message) error {
	40	+	item, ok := s.db.Get(ctx, []byte(key))
	41	+	if !ok {
	42	+	return domain.ErrNotFound
	43	+	}
	44	+	if err := proto.Unmarshal(item.Value, pb); err != nil {
	45	+	return fmt.Errorf("decode mirror record: %w", err)
	46	+	}
	47	+	return nil
	48	+	}
	49	+
	50	+	func (s *Store) writeMirror(ctx context.Context, key string, repo domain.RepoName, pb proto.Message) error {
	51	+	val := mustMarshal(pb)
	52	+	for range updateAttempts {
	53	+	var b kv.Batch
	54	+	if key != mirrorConfigKey {
	55	+	if err := guardRepoKey(ctx, s.db, &b, repo); err != nil {
	56	+	return err
	57	+	}
	58	+	}
	59	+	b.Put([]byte(key), val)
	60	+	_, err := s.db.Apply(ctx, &b)
	61	+	if errors.Is(err, kv.ErrConflict) {
	62	+	continue
	63	+	}
	64	+	if err != nil {
	65	+	return fmt.Errorf("store mirror record: %w", err)
	66	+	}
	67	+	return nil
	68	+	}
	69	+	return domain.ErrConflict
	70	+	}
	71	+
	72	+	func (s *Store) GetGlobal(ctx context.Context) (domain.MirrorGlobalConfig, error) {
	73	+	var pb storagepb.MirrorGlobal
	74	+	err := s.readMirror(ctx, mirrorConfigKey, &pb)
	75	+	if errors.Is(err, domain.ErrNotFound) {
	76	+	err = nil
	77	+	}
	78	+	return domain.MirrorGlobalConfig{Enabled: pb.Enabled, URLTemplate: pb.UrlTemplate, Cron: pb.Cron, AllBranches: pb.AllBranches, UpdatedAt: decodeTime(pb.UpdatedAt)}, err
	79	+	}
	80	+	func (s *Store) PutGlobal(ctx context.Context, g domain.MirrorGlobalConfig) error {
	81	+	return s.writeMirror(ctx, mirrorConfigKey, domain.RepoName{}, &storagepb.MirrorGlobal{Enabled: g.Enabled, UrlTemplate: g.URLTemplate, Cron: g.Cron, AllBranches: g.AllBranches, UpdatedAt: encodeTime(g.UpdatedAt)})
	82	+	}
	83	+	func decodeMirrorRepo(repo domain.RepoName, pb *storagepb.MirrorRepo) domain.MirrorRepoConfig {
	84	+	return domain.MirrorRepoConfig{Repo: repo, URL: pb.Url, Cron: pb.Cron, AllBranches: pb.AllBranches, Excluded: pb.Excluded, UpdatedAt: decodeTime(pb.UpdatedAt)}
	85	+	}
	86	+	func (s *Store) GetRepoConfig(ctx context.Context, repo domain.RepoName) (domain.MirrorRepoConfig, error) {
	87	+	var pb storagepb.MirrorRepo
	88	+	err := s.readMirror(ctx, mirrorRepoPrefix+repo.String(), &pb)
	89	+	return decodeMirrorRepo(repo, &pb), err
	90	+	}
	91	+	func (s *Store) PutRepoConfig(ctx context.Context, c domain.MirrorRepoConfig) error {
	92	+	return s.writeMirror(ctx, mirrorRepoPrefix+c.Repo.String(), c.Repo, &storagepb.MirrorRepo{Url: c.URL, Cron: c.Cron, AllBranches: c.AllBranches, Excluded: c.Excluded, UpdatedAt: encodeTime(c.UpdatedAt)})
	93	+	}
	94	+	func (s *Store) DeleteRepoConfig(ctx context.Context, repo domain.RepoName) error {
	95	+	var b kv.Batch
	96	+	b.Delete([]byte(mirrorRepoPrefix + repo.String()))
	97	+	_, err := s.db.Apply(ctx, &b)
	98	+	if err != nil {
	99	+	return fmt.Errorf("delete mirror config: %w", err)
	100	+	}
	101	+	return nil
	102	+	}
	103	+	func (s *Store) ListRepoConfigs(ctx context.Context) ([]domain.MirrorRepoConfig, error) {
	104	+	var out []domain.MirrorRepoConfig
	105	+	for key, item := range s.db.Snapshot(ctx).Prefix(ctx, []byte(mirrorRepoPrefix)) {
	106	+	repo, err := domain.ParseRepoName(strings.TrimPrefix(string(key), mirrorRepoPrefix))
	107	+	if err != nil {
	108	+	return nil, err
	109	+	}
	110	+	var pb storagepb.MirrorRepo
	111	+	if err := proto.Unmarshal(item.Value, &pb); err != nil {
	112	+	return nil, fmt.Errorf("decode mirror config: %w", err)
	113	+	}
	114	+	out = append(out, decodeMirrorRepo(repo, &pb))
	115	+	}
	116	+	return out, nil
	117	+	}
	118	+	func decodeMirrorStatus(repo domain.RepoName, pb *storagepb.MirrorState) domain.MirrorStatus {
	119	+	return domain.MirrorStatus{Repo: repo, State: domain.MirrorState(pb.State), Class: domain.MirrorFailureClass(pb.Class), Reason: pb.Reason, Detail: pb.Detail, ConsecutiveFailures: int(pb.ConsecutiveFailures), LastAttemptAt: decodeTime(pb.LastAttemptAt), LastSuccessAt: decodeTime(pb.LastSuccessAt), NextRetryAt: decodeTime(pb.NextRetryAt), PausedAt: decodeTime(pb.PausedAt), UpdatedAt: decodeTime(pb.UpdatedAt)}
	120	+	}
	121	+	func (s *Store) GetStatus(ctx context.Context, repo domain.RepoName) (domain.MirrorStatus, error) {
	122	+	var pb storagepb.MirrorState
	123	+	err := s.readMirror(ctx, mirrorStatePrefix+repo.String(), &pb)
	124	+	return decodeMirrorStatus(repo, &pb), err
	125	+	}
	126	+	func (s *Store) PutStatus(ctx context.Context, c domain.MirrorStatus) error {
	127	+	return s.writeMirror(ctx, mirrorStatePrefix+c.Repo.String(), c.Repo, &storagepb.MirrorState{State: string(c.State), Class: string(c.Class), Reason: c.Reason, Detail: c.Detail, ConsecutiveFailures: int32(c.ConsecutiveFailures), LastAttemptAt: encodeTime(c.LastAttemptAt), LastSuccessAt: encodeTime(c.LastSuccessAt), NextRetryAt: encodeTime(c.NextRetryAt), PausedAt: encodeTime(c.PausedAt), UpdatedAt: encodeTime(c.UpdatedAt)})
	128	+	}
	129	+	func (s *Store) ListStatus(ctx context.Context) ([]domain.MirrorStatus, error) {
	130	+	var out []domain.MirrorStatus
	131	+	for key, item := range s.db.Snapshot(ctx).Prefix(ctx, []byte(mirrorStatePrefix)) {
	132	+	repo, err := domain.ParseRepoName(strings.TrimPrefix(string(key), mirrorStatePrefix))
	133	+	if err != nil {
	134	+	return nil, err
	135	+	}
	136	+	var pb storagepb.MirrorState
	137	+	if err := proto.Unmarshal(item.Value, &pb); err != nil {
	138	+	return nil, fmt.Errorf("decode mirror state: %w", err)
	139	+	}
	140	+	out = append(out, decodeMirrorStatus(repo, &pb))
	141	+	}
	142	+	return out, nil
	143	+	}
	144	+	func (s *Store) GetSnapshot(ctx context.Context, repo domain.RepoName) (domain.MirrorSnapshot, error) {
	145	+	var pb storagepb.MirrorSnapshot
	146	+	err := s.readMirror(ctx, mirrorSnapshotPrefix+repo.String(), &pb)
	147	+	out := domain.MirrorSnapshot{URL: pb.Url, AllBranches: pb.AllBranches, At: decodeTime(pb.At)}
	148	+	for _, ref := range pb.Refs {
	149	+	id, parseErr := domain.ParseObjectID(ref.Id)
	150	+	if parseErr != nil {
	151	+	return domain.MirrorSnapshot{}, fmt.Errorf("decode mirror snapshot ref: %w", parseErr)
	152	+	}
	153	+	out.Refs = append(out.Refs, domain.MirrorRef{Name: ref.Name, ID: id})
	154	+	}
	155	+	return out, err
	156	+	}
	157	+	func (s *Store) PutSnapshot(ctx context.Context, repo domain.RepoName, c domain.MirrorSnapshot) error {
	158	+	pb := &storagepb.MirrorSnapshot{Url: c.URL, AllBranches: c.AllBranches, At: encodeTime(c.At)}
	159	+	for _, ref := range c.Refs {
	160	+	pb.Refs = append(pb.Refs, &storagepb.MirrorSnapshotRef{Name: ref.Name, Id: ref.ID.String()})
	161	+	}
	162	+	return s.writeMirror(ctx, mirrorSnapshotPrefix+repo.String(), repo, pb)
	163	+	}
	164	+	func (s *Store) GetMirrorKey(ctx context.Context) ([]byte, time.Time, error) {
	165	+	var pb storagepb.MirrorKey
	166	+	err := s.readMirror(ctx, domain.MirrorKeyPath, &pb)
	167	+	return pb.Sealed, decodeTime(pb.CreatedAt), err
	168	+	}
	169	+	func (s *Store) PutMirrorKey(ctx context.Context, sealed []byte, at time.Time, ifAbsent bool) error {
	170	+	var b kv.Batch
	171	+	key := []byte(domain.MirrorKeyPath)
	172	+	if ifAbsent {
	173	+	b.RequireAbsent(key)
	174	+	}
	175	+	b.Put(key, mustMarshal(&storagepb.MirrorKey{Sealed: sealed, CreatedAt: encodeTime(at)}))
	176	+	_, err := s.db.Apply(ctx, &b)
	177	+	if ifAbsent && errors.Is(err, kv.ErrConflict) {
	178	+	return domain.ErrAlreadyExists
	179	+	}
	180	+	if err != nil {
	181	+	return fmt.Errorf("store mirror key: %w", err)
	182	+	}
	183	+	return nil
	184	+	}
	185	+
	186	+	// RawMirrorKey exposes the stored protobuf for encryption-at-rest tests.
	187	+	func (s *Store) RawMirrorKey(ctx context.Context) ([]byte, bool) {
	188	+	item, ok := s.db.Get(ctx, []byte(domain.MirrorKeyPath))
	189	+	return item.Value, ok
	190	+	}

A internal/adapters/kvstore/mirror_test.go +98 -0

@@ -0,0 +1,98 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package kvstore_test
	4	+
	5	+	import (
	6	+	"bytes"
	7	+	"context"
	8	+	"testing"
	9	+
	10	+	"github.com/stretchr/testify/require"
	11	+
	12	+	"bnema.dev/ermite/internal/domain"
	13	+	)
	14	+
	15	+	func TestMirrorRoundTripsAndDelete(t *testing.T) {
	16	+	ctx := context.Background()
	17	+	s := openStore(t, t.TempDir())
	18	+	repo := name(t, "mirror")
	19	+	g, err := s.GetGlobal(ctx)
	20	+	require.NoError(t, err)
	21	+	require.Equal(t, domain.MirrorGlobalConfig{}, g)
	22	+	_, err = s.GetRepoConfig(ctx, repo)
	23	+	require.ErrorIs(t, err, domain.ErrNotFound)
	24	+	_, err = s.GetStatus(ctx, repo)
	25	+	require.ErrorIs(t, err, domain.ErrNotFound)
	26	+	_, err = s.GetSnapshot(ctx, repo)
	27	+	require.ErrorIs(t, err, domain.ErrNotFound)
	28	+	_, _, err = s.GetMirrorKey(ctx)
	29	+	require.ErrorIs(t, err, domain.ErrNotFound)
	30	+	require.NoError(t, s.DeleteRepoConfig(ctx, repo))
	31	+	require.ErrorIs(t, s.PutRepoConfig(ctx, domain.MirrorRepoConfig{Repo: repo}), domain.ErrNotFound)
	32	+	require.ErrorIs(t, s.PutStatus(ctx, domain.MirrorStatus{Repo: repo}), domain.ErrNotFound)
	33	+	require.ErrorIs(t, s.PutSnapshot(ctx, repo, domain.MirrorSnapshot{}), domain.ErrNotFound)
	34	+	require.NoError(t, s.Create(ctx, domain.Repo{Name: repo, Visibility: domain.VisibilityPrivate}))
	35	+	g = domain.MirrorGlobalConfig{Enabled: true, URLTemplate: "git@example.org:{repo}", Cron: "/5 * * *", AllBranches: true, UpdatedAt: created}
	36	+	require.NoError(t, s.PutGlobal(ctx, g))
	37	+	gg, err := s.GetGlobal(ctx)
	38	+	require.NoError(t, err)
	39	+	require.Equal(t, g, gg)
	40	+	all := false
	41	+	c := domain.MirrorRepoConfig{Repo: repo, URL: "git@example.org:override", Cron: "0 * * * *", AllBranches: &all, Excluded: true, UpdatedAt: created}
	42	+	require.NoError(t, s.PutRepoConfig(ctx, c))
	43	+	cc, err := s.GetRepoConfig(ctx, repo)
	44	+	require.NoError(t, err)
	45	+	require.Equal(t, c, cc)
	46	+	cs, err := s.ListRepoConfigs(ctx)
	47	+	require.NoError(t, err)
	48	+	require.Equal(t, []domain.MirrorRepoConfig{c}, cs)
	49	+	c.AllBranches = nil
	50	+	require.NoError(t, s.PutRepoConfig(ctx, c))
	51	+	cc, err = s.GetRepoConfig(ctx, repo)
	52	+	require.NoError(t, err)
	53	+	require.Nil(t, cc.AllBranches)
	54	+	st := domain.MirrorStatus{Repo: repo, State: domain.MirrorFailing, Class: domain.MirrorClassTransient, Reason: "retry", Detail: "network", ConsecutiveFailures: 3, LastAttemptAt: created, LastSuccessAt: created, NextRetryAt: created, PausedAt: created, UpdatedAt: created}
	55	+	require.NoError(t, s.PutStatus(ctx, st))
	56	+	ss, err := s.GetStatus(ctx, repo)
	57	+	require.NoError(t, err)
	58	+	require.Equal(t, st, ss)
	59	+	statuses, err := s.ListStatus(ctx)
	60	+	require.NoError(t, err)
	61	+	require.Equal(t, []domain.MirrorStatus{st}, statuses)
	62	+	id, err := domain.ParseObjectID("0123456789012345678901234567890123456789")
	63	+	require.NoError(t, err)
	64	+	snap := domain.MirrorSnapshot{URL: c.URL, AllBranches: true, At: created, Refs: []domain.MirrorRef{{Name: "refs/heads/main", ID: id}}}
	65	+	require.NoError(t, s.PutSnapshot(ctx, repo, snap))
	66	+	got, err := s.GetSnapshot(ctx, repo)
	67	+	require.NoError(t, err)
	68	+	require.Equal(t, snap, got)
	69	+	sealed := []byte("opaque sealed seed")
	70	+	require.NoError(t, s.PutMirrorKey(ctx, sealed, created, true))
	71	+	require.ErrorIs(t, s.PutMirrorKey(ctx, []byte("other"), created, true), domain.ErrAlreadyExists)
	72	+	blob, at, err := s.GetMirrorKey(ctx)
	73	+	require.NoError(t, err)
	74	+	require.Equal(t, sealed, blob)
	75	+	require.Equal(t, created, at)
	76	+	raw, ok := s.RawMirrorKey(ctx)
	77	+	require.True(t, ok)
	78	+	require.False(t, bytes.Equal(raw, sealed), "raw record is a protobuf envelope, not bare key material")
	79	+	require.NoError(t, s.PutMirrorKey(ctx, []byte("rotated"), created, false))
	80	+	blob, _, err = s.GetMirrorKey(ctx)
	81	+	require.NoError(t, err)
	82	+	require.Equal(t, []byte("rotated"), blob)
	83	+	require.NoError(t, s.Delete(ctx, repo))
	84	+	_, err = s.GetRepoConfig(ctx, repo)
	85	+	require.ErrorIs(t, err, domain.ErrNotFound)
	86	+	_, err = s.GetStatus(ctx, repo)
	87	+	require.ErrorIs(t, err, domain.ErrNotFound)
	88	+	_, err = s.GetSnapshot(ctx, repo)
	89	+	require.ErrorIs(t, err, domain.ErrNotFound)
	90	+	_, _, err = s.GetMirrorKey(ctx)
	91	+	require.NoError(t, err)
	92	+	gg, err = s.GetGlobal(ctx)
	93	+	require.NoError(t, err)
	94	+	require.Equal(t, g, gg)
	95	+	require.NoError(t, s.Create(ctx, domain.Repo{Name: repo, Visibility: domain.VisibilityPrivate}))
	96	+	_, err = s.GetStatus(ctx, repo)
	97	+	require.ErrorIs(t, err, domain.ErrNotFound)
	98	+	}

@@ -0,0 +1,98 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package kvstore_test
	4	+
	5	+	import (
	6	+	"bytes"
	7	+	"context"
	8	+	"testing"
	9	+
	10	+	"github.com/stretchr/testify/require"
	11	+
	12	+	"bnema.dev/ermite/internal/domain"
	13	+	)
	14	+
	15	+	func TestMirrorRoundTripsAndDelete(t *testing.T) {
	16	+	ctx := context.Background()
	17	+	s := openStore(t, t.TempDir())
	18	+	repo := name(t, "mirror")
	19	+	g, err := s.GetGlobal(ctx)
	20	+	require.NoError(t, err)
	21	+	require.Equal(t, domain.MirrorGlobalConfig{}, g)
	22	+	_, err = s.GetRepoConfig(ctx, repo)
	23	+	require.ErrorIs(t, err, domain.ErrNotFound)
	24	+	_, err = s.GetStatus(ctx, repo)
	25	+	require.ErrorIs(t, err, domain.ErrNotFound)
	26	+	_, err = s.GetSnapshot(ctx, repo)
	27	+	require.ErrorIs(t, err, domain.ErrNotFound)
	28	+	_, _, err = s.GetMirrorKey(ctx)
	29	+	require.ErrorIs(t, err, domain.ErrNotFound)
	30	+	require.NoError(t, s.DeleteRepoConfig(ctx, repo))
	31	+	require.ErrorIs(t, s.PutRepoConfig(ctx, domain.MirrorRepoConfig{Repo: repo}), domain.ErrNotFound)
	32	+	require.ErrorIs(t, s.PutStatus(ctx, domain.MirrorStatus{Repo: repo}), domain.ErrNotFound)
	33	+	require.ErrorIs(t, s.PutSnapshot(ctx, repo, domain.MirrorSnapshot{}), domain.ErrNotFound)
	34	+	require.NoError(t, s.Create(ctx, domain.Repo{Name: repo, Visibility: domain.VisibilityPrivate}))
	35	+	g = domain.MirrorGlobalConfig{Enabled: true, URLTemplate: "git@example.org:{repo}", Cron: "/5 * * *", AllBranches: true, UpdatedAt: created}
	36	+	require.NoError(t, s.PutGlobal(ctx, g))
	37	+	gg, err := s.GetGlobal(ctx)
	38	+	require.NoError(t, err)
	39	+	require.Equal(t, g, gg)
	40	+	all := false
	41	+	c := domain.MirrorRepoConfig{Repo: repo, URL: "git@example.org:override", Cron: "0 * * * *", AllBranches: &all, Excluded: true, UpdatedAt: created}
	42	+	require.NoError(t, s.PutRepoConfig(ctx, c))
	43	+	cc, err := s.GetRepoConfig(ctx, repo)
	44	+	require.NoError(t, err)
	45	+	require.Equal(t, c, cc)
	46	+	cs, err := s.ListRepoConfigs(ctx)
	47	+	require.NoError(t, err)
	48	+	require.Equal(t, []domain.MirrorRepoConfig{c}, cs)
	49	+	c.AllBranches = nil
	50	+	require.NoError(t, s.PutRepoConfig(ctx, c))
	51	+	cc, err = s.GetRepoConfig(ctx, repo)
	52	+	require.NoError(t, err)
	53	+	require.Nil(t, cc.AllBranches)
	54	+	st := domain.MirrorStatus{Repo: repo, State: domain.MirrorFailing, Class: domain.MirrorClassTransient, Reason: "retry", Detail: "network", ConsecutiveFailures: 3, LastAttemptAt: created, LastSuccessAt: created, NextRetryAt: created, PausedAt: created, UpdatedAt: created}
	55	+	require.NoError(t, s.PutStatus(ctx, st))
	56	+	ss, err := s.GetStatus(ctx, repo)
	57	+	require.NoError(t, err)
	58	+	require.Equal(t, st, ss)
	59	+	statuses, err := s.ListStatus(ctx)
	60	+	require.NoError(t, err)
	61	+	require.Equal(t, []domain.MirrorStatus{st}, statuses)
	62	+	id, err := domain.ParseObjectID("0123456789012345678901234567890123456789")
	63	+	require.NoError(t, err)
	64	+	snap := domain.MirrorSnapshot{URL: c.URL, AllBranches: true, At: created, Refs: []domain.MirrorRef{{Name: "refs/heads/main", ID: id}}}
	65	+	require.NoError(t, s.PutSnapshot(ctx, repo, snap))
	66	+	got, err := s.GetSnapshot(ctx, repo)
	67	+	require.NoError(t, err)
	68	+	require.Equal(t, snap, got)
	69	+	sealed := []byte("opaque sealed seed")
	70	+	require.NoError(t, s.PutMirrorKey(ctx, sealed, created, true))
	71	+	require.ErrorIs(t, s.PutMirrorKey(ctx, []byte("other"), created, true), domain.ErrAlreadyExists)
	72	+	blob, at, err := s.GetMirrorKey(ctx)
	73	+	require.NoError(t, err)
	74	+	require.Equal(t, sealed, blob)
	75	+	require.Equal(t, created, at)
	76	+	raw, ok := s.RawMirrorKey(ctx)
	77	+	require.True(t, ok)
	78	+	require.False(t, bytes.Equal(raw, sealed), "raw record is a protobuf envelope, not bare key material")
	79	+	require.NoError(t, s.PutMirrorKey(ctx, []byte("rotated"), created, false))
	80	+	blob, _, err = s.GetMirrorKey(ctx)
	81	+	require.NoError(t, err)
	82	+	require.Equal(t, []byte("rotated"), blob)
	83	+	require.NoError(t, s.Delete(ctx, repo))
	84	+	_, err = s.GetRepoConfig(ctx, repo)
	85	+	require.ErrorIs(t, err, domain.ErrNotFound)
	86	+	_, err = s.GetStatus(ctx, repo)
	87	+	require.ErrorIs(t, err, domain.ErrNotFound)
	88	+	_, err = s.GetSnapshot(ctx, repo)
	89	+	require.ErrorIs(t, err, domain.ErrNotFound)
	90	+	_, _, err = s.GetMirrorKey(ctx)
	91	+	require.NoError(t, err)
	92	+	gg, err = s.GetGlobal(ctx)
	93	+	require.NoError(t, err)
	94	+	require.Equal(t, g, gg)
	95	+	require.NoError(t, s.Create(ctx, domain.Repo{Name: repo, Visibility: domain.VisibilityPrivate}))
	96	+	_, err = s.GetStatus(ctx, repo)
	97	+	require.ErrorIs(t, err, domain.ErrNotFound)
	98	+	}

M internal/adapters/kvstore/repos.go +1 -0


@@ -164,6 +164,7 @@ func (s *Store) Delete(ctx context.Context, name domain.RepoName) error {
164	164		keys = append(keys, pendingKeys(ctx, snap, name)...)
165	165		keys = append(keys, ciKeys(ctx, snap, name)...)
166	166		keys = append(keys, settingsKeys(ctx, snap, name)...)
	167	+	keys = append(keys, mirrorKeys(ctx, snap, name)...)
167	168		for len(keys) > deleteChunk {
168	169		var b kv.Batch
169	170		for _, k := range keys[:deleteChunk] {


@@ -164,6 +164,7 @@ func (s *Store) Delete(ctx context.Context, name domain.RepoName) error {
164	keys = append(keys, pendingKeys(ctx, snap, name)...)	164		keys = append(keys, pendingKeys(ctx, snap, name)...)
165	keys = append(keys, ciKeys(ctx, snap, name)...)	165		keys = append(keys, ciKeys(ctx, snap, name)...)
166	keys = append(keys, settingsKeys(ctx, snap, name)...)	166		keys = append(keys, settingsKeys(ctx, snap, name)...)
		167	+	keys = append(keys, mirrorKeys(ctx, snap, name)...)
167	for len(keys) > deleteChunk {	168		for len(keys) > deleteChunk {
168	var b kv.Batch	169		var b kv.Batch
169	for _, k := range keys[:deleteChunk] {	170		for _, k := range keys[:deleteChunk] {

A proto/ermite/storage/mirror.proto +45 -0

@@ -0,0 +1,45 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+	syntax = "proto3";
	3	+	package ermite.storage;
	4	+	import "google/protobuf/timestamp.proto";
	5	+	option go_package = "bnema.dev/ermite/gen/ermite/storage;storagepb";
	6	+	message MirrorKey {
	7	+	bytes sealed = 1;
	8	+	google.protobuf.Timestamp created_at = 2;
	9	+	}
	10	+	message MirrorGlobal {
	11	+	bool enabled = 1;
	12	+	string url_template = 2;
	13	+	string cron = 3;
	14	+	bool all_branches = 4;
	15	+	google.protobuf.Timestamp updated_at = 5;
	16	+	}
	17	+	message MirrorRepo {
	18	+	string url = 1;
	19	+	string cron = 2;
	20	+	optional bool all_branches = 3;
	21	+	bool excluded = 4;
	22	+	google.protobuf.Timestamp updated_at = 5;
	23	+	}
	24	+	message MirrorState {
	25	+	string state = 1;
	26	+	string class = 2;
	27	+	string reason = 3;
	28	+	string detail = 4;
	29	+	int32 consecutive_failures = 5;
	30	+	google.protobuf.Timestamp last_attempt_at = 6;
	31	+	google.protobuf.Timestamp last_success_at = 7;
	32	+	google.protobuf.Timestamp next_retry_at = 8;
	33	+	google.protobuf.Timestamp paused_at = 9;
	34	+	google.protobuf.Timestamp updated_at = 10;
	35	+	}
	36	+	message MirrorSnapshotRef {
	37	+	string name = 1;
	38	+	string id = 2;
	39	+	}
	40	+	message MirrorSnapshot {
	41	+	string url = 1;
	42	+	repeated MirrorSnapshotRef refs = 2;
	43	+	bool all_branches = 3;
	44	+	google.protobuf.Timestamp at = 4;
	45	+	}

@@ -0,0 +1,45 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+	syntax = "proto3";
	3	+	package ermite.storage;
	4	+	import "google/protobuf/timestamp.proto";
	5	+	option go_package = "bnema.dev/ermite/gen/ermite/storage;storagepb";
	6	+	message MirrorKey {
	7	+	bytes sealed = 1;
	8	+	google.protobuf.Timestamp created_at = 2;
	9	+	}
	10	+	message MirrorGlobal {
	11	+	bool enabled = 1;
	12	+	string url_template = 2;
	13	+	string cron = 3;
	14	+	bool all_branches = 4;
	15	+	google.protobuf.Timestamp updated_at = 5;
	16	+	}
	17	+	message MirrorRepo {
	18	+	string url = 1;
	19	+	string cron = 2;
	20	+	optional bool all_branches = 3;
	21	+	bool excluded = 4;
	22	+	google.protobuf.Timestamp updated_at = 5;
	23	+	}
	24	+	message MirrorState {
	25	+	string state = 1;
	26	+	string class = 2;
	27	+	string reason = 3;
	28	+	string detail = 4;
	29	+	int32 consecutive_failures = 5;
	30	+	google.protobuf.Timestamp last_attempt_at = 6;
	31	+	google.protobuf.Timestamp last_success_at = 7;
	32	+	google.protobuf.Timestamp next_retry_at = 8;
	33	+	google.protobuf.Timestamp paused_at = 9;
	34	+	google.protobuf.Timestamp updated_at = 10;
	35	+	}
	36	+	message MirrorSnapshotRef {
	37	+	string name = 1;
	38	+	string id = 2;
	39	+	}
	40	+	message MirrorSnapshot {
	41	+	string url = 1;
	42	+	repeated MirrorSnapshotRef refs = 2;
	43	+	bool all_branches = 3;
	44	+	google.protobuf.Timestamp at = 4;
	45	+	}