638943e Merge pull request #90 from bnema/fix/registry-shutdown-resilience · gordon

M internal/app/migrate_env_test.go +6 -3


@@ -110,11 +110,14 @@ func TestMigrateAttachmentEnvFile(t *testing.T) {
110	110		require.NoError(t, err)
111	111
112	112		containerName := fmt.Sprintf("gitea-postgres-%d", time.Now().UnixNano())
	113	+	envFileName := "gordon-" + containerName + ".env"
	114	+	// extractContainerNameFromAttachmentFile strips the .env suffix, so the
	115	+	// stored name includes the "gordon-" prefix from the filename.
	116	+	storedContainerName := extractContainerNameFromAttachmentFile(envFileName)
113	117		keys := []string{"POSTGRES_USER", "POSTGRES_PASSWORD"}
114		-	defer cleanupPassAttachment(t, containerName, keys)
	118	+	defer cleanupPassAttachment(t, storedContainerName, keys)
115	119
116	120		envContent := "POSTGRES_USER=gitea\nPOSTGRES_PASSWORD=secret123\n"
117		-	envFileName := "gordon-" + containerName + ".env"
118	121		envFilePath := filepath.Join(tmpDir, envFileName)
119	122		err = os.WriteFile(envFilePath, []byte(envContent), 0600)
120	123		require.NoError(t, err)

@@ -129,7 +132,7 @@ func TestMigrateAttachmentEnvFile(t *testing.T) {
129	132		_, err = os.Stat(migratedPath)
130	133		assert.NoError(t, err, ".env.migrated file should exist")
131	134
132		-	values, err := store.GetAllAttachment(containerName)
	135	+	values, err := store.GetAllAttachment(storedContainerName)
133	136		require.NoError(t, err)
134	137		assert.Equal(t, "gitea", values["POSTGRES_USER"])
135	138		assert.Equal(t, "secret123", values["POSTGRES_PASSWORD"])

 	require.NoError(t, err)
 	containerName := fmt.Sprintf("gitea-postgres-%d", time.Now().UnixNano())
+	envFileName := "gordon-" + containerName + ".env"
+	// extractContainerNameFromAttachmentFile strips the .env suffix, so the
+	// stored name includes the "gordon-" prefix from the filename.
+	storedContainerName := extractContainerNameFromAttachmentFile(envFileName)
 	keys := []string{"POSTGRES_USER", "POSTGRES_PASSWORD"}
-	defer cleanupPassAttachment(t, containerName, keys)
+	defer cleanupPassAttachment(t, storedContainerName, keys)
 	envContent := "POSTGRES_USER=gitea\nPOSTGRES_PASSWORD=secret123\n"
-	envFileName := "gordon-" + containerName + ".env"
 	envFilePath := filepath.Join(tmpDir, envFileName)
 	err = os.WriteFile(envFilePath, []byte(envContent), 0600)
 	require.NoError(t, err)
 	_, err = os.Stat(migratedPath)
 	assert.NoError(t, err, ".env.migrated file should exist")
-	values, err := store.GetAllAttachment(containerName)
+	values, err := store.GetAllAttachment(storedContainerName)
 	require.NoError(t, err)
 	assert.Equal(t, "gitea", values["POSTGRES_USER"])
 	assert.Equal(t, "secret123", values["POSTGRES_PASSWORD"])

M internal/app/run.go +19 -3


@@ -1750,7 +1750,7 @@ func runServers(ctx context.Context, v *viper.Viper, cfg Config, registryHandler
1750	1750		syncAndAutoStart(ctx, svc, log)
1751	1751
1752	1752		waitForShutdown(ctx, errChan, reloadChan, deployChan, eventBus, log)
1753		-	gracefulShutdown(registrySrv, proxySrv, tlsSrv, containerSvc, log)
	1753	+	gracefulShutdown(registrySrv, proxySrv, tlsSrv, containerSvc, svc.proxySvc, log)
1754	1754		return nil
1755	1755		}
1756	1756

@@ -1950,13 +1950,14 @@ func waitForShutdown(ctx context.Context, errChan <-chan error, reloadChan, depl
1950	1950
1951	1951		// gracefulShutdown stops HTTP servers with a 30s timeout, then shuts down
1952	1952		// the container service and cleans up runtime files.
1953		-	func gracefulShutdown(registrySrv, proxySrv, tlsSrv http.Server, containerSvc container.Service, log zerowrap.Logger) {
	1953	+	func gracefulShutdown(registrySrv, proxySrv, tlsSrv http.Server, containerSvc container.Service, proxySvc *proxy.Service, log zerowrap.Logger) {
1954	1954		log.Info().Msg("shutting down Gordon...")
1955	1955
1956	1956		shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 30*time.Second)
1957	1957		defer shutdownCancel()
1958	1958
1959		-	for _, srv := range []*http.Server{registrySrv, proxySrv, tlsSrv} {
	1959	+	// Phase 1: Stop ingress frontends (TLS, then proxy) — no new traffic accepted
	1960	+	for _, srv := range []*http.Server{tlsSrv, proxySrv} {
1960	1961		if srv == nil {
1961	1962		continue
1962	1963		}

@@ -1965,6 +1966,21 @@ func gracefulShutdown(registrySrv, proxySrv, tlsSrv http.Server, containerSvc
1965	1966		}
1966	1967		}
1967	1968
	1969	+	// Phase 2: Drain in-flight registry push sessions before stopping the backend
	1970	+	if proxySvc != nil {
	1971	+	log.Info().Msg("draining in-flight registry requests...")
	1972	+	if drained := proxySvc.DrainRegistryInFlight(25 * time.Second); !drained {
	1973	+	log.Warn().Int64("in_flight", proxySvc.RegistryInFlight()).Msg("registry drain timed out; some in-flight pushes may be interrupted")
	1974	+	}
	1975	+	}
	1976	+
	1977	+	// Phase 3: Stop the registry backend
	1978	+	if registrySrv != nil {
	1979	+	if err := registrySrv.Shutdown(shutdownCtx); err != nil {
	1980	+	log.Warn().Err(err).Str("addr", registrySrv.Addr).Msg("server shutdown error")
	1981	+	}
	1982	+	}
	1983	+
1968	1984		containerSvc.StopMonitor()
1969	1985
1970	1986		if err := containerSvc.Shutdown(shutdownCtx); err != nil {

 	syncAndAutoStart(ctx, svc, log)
 	waitForShutdown(ctx, errChan, reloadChan, deployChan, eventBus, log)
-	gracefulShutdown(registrySrv, proxySrv, tlsSrv, containerSvc, log)
+	gracefulShutdown(registrySrv, proxySrv, tlsSrv, containerSvc, svc.proxySvc, log)
 	return nil
 }
 // gracefulShutdown stops HTTP servers with a 30s timeout, then shuts down
 // the container service and cleans up runtime files.
-func gracefulShutdown(registrySrv, proxySrv, tlsSrv *http.Server, containerSvc *container.Service, log zerowrap.Logger) {
+func gracefulShutdown(registrySrv, proxySrv, tlsSrv *http.Server, containerSvc *container.Service, proxySvc *proxy.Service, log zerowrap.Logger) {
 	log.Info().Msg("shutting down Gordon...")
 	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer shutdownCancel()
-	for _, srv := range []*http.Server{registrySrv, proxySrv, tlsSrv} {
+	// Phase 1: Stop ingress frontends (TLS, then proxy) — no new traffic accepted
+	for _, srv := range []*http.Server{tlsSrv, proxySrv} {
 		if srv == nil {
 			continue
 		}
 		}
 	}
+	// Phase 2: Drain in-flight registry push sessions before stopping the backend
+	if proxySvc != nil {
+		log.Info().Msg("draining in-flight registry requests...")
+		if drained := proxySvc.DrainRegistryInFlight(25 * time.Second); !drained {
+			log.Warn().Int64("in_flight", proxySvc.RegistryInFlight()).Msg("registry drain timed out; some in-flight pushes may be interrupted")
+		}
+	}
+	// Phase 3: Stop the registry backend
+	if registrySrv != nil {
+		if err := registrySrv.Shutdown(shutdownCtx); err != nil {
+			log.Warn().Err(err).Str("addr", registrySrv.Addr).Msg("server shutdown error")
+		}
+	}
 	containerSvc.StopMonitor()
 	if err := containerSvc.Shutdown(shutdownCtx); err != nil {

M internal/usecase/proxy/service.go +33 -9


@@ -70,15 +70,16 @@ type Config struct {
70	70
71	71		// Service implements the ProxyService interface.
72	72		type Service struct {
73		-	runtime out.ContainerRuntime
74		-	containerSvc in.ContainerService
75		-	configSvc in.ConfigService
76		-	config Config
77		-	targets map[string]*domain.ProxyTarget
78		-	mu sync.RWMutex
79		-	activeConns atomic.Int64 // Atomic counter for concurrent connection limiting
80		-	inFlight map[string]int
81		-	inFlightMu sync.Mutex
	73	+	runtime out.ContainerRuntime
	74	+	containerSvc in.ContainerService
	75	+	configSvc in.ConfigService
	76	+	config Config
	77	+	targets map[string]*domain.ProxyTarget
	78	+	mu sync.RWMutex
	79	+	activeConns atomic.Int64 // Atomic counter for concurrent connection limiting
	80	+	inFlight map[string]int
	81	+	inFlightMu sync.Mutex
	82	+	registryInFlight atomic.Int64 // active registry proxy requests, for graceful drain
82	83		}
83	84
84	85		// NewService creates a new proxy service.

@@ -574,6 +575,26 @@ func (s *Service) trackInFlight(containerID string) func() {
574	575		}
575	576		}
576	577
	578	+	// RegistryInFlight returns the current count of active registry proxy requests.
	579	+	func (s *Service) RegistryInFlight() int64 {
	580	+	return s.registryInFlight.Load()
	581	+	}
	582	+
	583	+	// DrainRegistryInFlight blocks until all in-flight registry proxy requests
	584	+	// complete or the timeout elapses. Returns true if drained cleanly, false if
	585	+	// timed out with requests still in flight. Call this before shutting down the
	586	+	// registry server.
	587	+	func (s *Service) DrainRegistryInFlight(timeout time.Duration) bool {
	588	+	deadline := time.Now().Add(timeout)
	589	+	for time.Now().Before(deadline) {
	590	+	if s.registryInFlight.Load() == 0 {
	591	+	return true
	592	+	}
	593	+	time.Sleep(10 * time.Millisecond)
	594	+	}
	595	+	return false
	596	+	}
	597	+
577	598		// proxyToRegistry forwards requests to the local registry HTTP server.
578	599		// SECURITY: This uses http://localhost:{port} which is safe because the registry
579	600		// runs on the same host and traffic never leaves the loopback interface. If Gordon

@@ -585,6 +606,9 @@ func (s Service) proxyToRegistry(w http.ResponseWriter, r http.Request) {
585	606		cfg := s.config
586	607		s.mu.RUnlock()
587	608
	609	+	s.registryInFlight.Add(1)
	610	+	defer s.registryInFlight.Add(-1)
	611	+
588	612		log := zerowrap.FromCtx(r.Context())
589	613
590	614		targetURL, err := url.Parse(fmt.Sprintf("http://localhost:%d", cfg.RegistryPort))

 // Service implements the ProxyService interface.
 type Service struct {
-	runtime      out.ContainerRuntime
+	runtime          out.ContainerRuntime
-	containerSvc in.ContainerService
+	containerSvc     in.ContainerService
-	configSvc    in.ConfigService
+	configSvc        in.ConfigService
-	config       Config
+	config           Config
-	targets      map[string]*domain.ProxyTarget
+	targets          map[string]*domain.ProxyTarget
-	mu           sync.RWMutex
+	mu               sync.RWMutex
-	activeConns  atomic.Int64 // Atomic counter for concurrent connection limiting
+	activeConns      atomic.Int64 // Atomic counter for concurrent connection limiting
-	inFlight     map[string]int
+	inFlight         map[string]int
-	inFlightMu   sync.Mutex
+	inFlightMu       sync.Mutex
+	registryInFlight atomic.Int64 // active registry proxy requests, for graceful drain
 }
 // NewService creates a new proxy service.
 	}
 }
+// RegistryInFlight returns the current count of active registry proxy requests.
+func (s *Service) RegistryInFlight() int64 {
+	return s.registryInFlight.Load()
+}
+// DrainRegistryInFlight blocks until all in-flight registry proxy requests
+// complete or the timeout elapses. Returns true if drained cleanly, false if
+// timed out with requests still in flight. Call this before shutting down the
+// registry server.
+func (s *Service) DrainRegistryInFlight(timeout time.Duration) bool {
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		if s.registryInFlight.Load() == 0 {
+			return true
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	return false
+}
 // proxyToRegistry forwards requests to the local registry HTTP server.
 // SECURITY: This uses http://localhost:{port} which is safe because the registry
 // runs on the same host and traffic never leaves the loopback interface. If Gordon
 	cfg := s.config
 	s.mu.RUnlock()
+	s.registryInFlight.Add(1)
+	defer s.registryInFlight.Add(-1)
 	log := zerowrap.FromCtx(r.Context())
 	targetURL, err := url.Parse(fmt.Sprintf("http://localhost:%d", cfg.RegistryPort))

M internal/usecase/proxy/service_test.go +64 -0


@@ -4,6 +4,7 @@ import (
4	4		"context"
5	5		"net/http/httptest"
6	6		"testing"
	7	+	"time"
7	8
8	9		"github.com/bnema/zerowrap"
9	10		"github.com/stretchr/testify/assert"

@@ -510,3 +511,66 @@ func TestServeHTTP_MaxBodySize(t *testing.T) {
510	511		})
511	512		}
512	513		}
	514	+
	515	+	func TestRegistryInFlightTracking(t *testing.T) {
	516	+	svc := &Service{
	517	+	inFlight: make(map[string]int),
	518	+	}
	519	+
	520	+	if got := svc.registryInFlight.Load(); got != 0 {
	521	+	t.Fatalf("expected 0 in-flight, got %d", got)
	522	+	}
	523	+
	524	+	svc.registryInFlight.Add(1)
	525	+	if got := svc.registryInFlight.Load(); got != 1 {
	526	+	t.Fatalf("expected 1 in-flight after Add, got %d", got)
	527	+	}
	528	+
	529	+	svc.registryInFlight.Add(-1)
	530	+	if got := svc.registryInFlight.Load(); got != 0 {
	531	+	t.Fatalf("expected 0 in-flight after release, got %d", got)
	532	+	}
	533	+	}
	534	+
	535	+	func TestDrainRegistryInFlight(t *testing.T) {
	536	+	svc := &Service{
	537	+	inFlight: make(map[string]int),
	538	+	}
	539	+
	540	+	svc.registryInFlight.Add(2)
	541	+
	542	+	done := make(chan struct{})
	543	+	go func() {
	544	+	drained := svc.DrainRegistryInFlight(50 * time.Millisecond)
	545	+	if !drained {
	546	+	// Signal failure via done channel by leaving it open — test will time out
	547	+	return
	548	+	}
	549	+	close(done)
	550	+	}()
	551	+
	552	+	time.Sleep(5 * time.Millisecond)
	553	+	svc.registryInFlight.Add(-1)
	554	+	svc.registryInFlight.Add(-1)
	555	+
	556	+	select {
	557	+	case <-done:
	558	+	// good — drained cleanly
	559	+	case <-time.After(500 * time.Millisecond):
	560	+	t.Fatal("DrainRegistryInFlight did not return true after requests completed")
	561	+	}
	562	+	}
	563	+
	564	+	func TestDrainRegistryInFlightTimeout(t *testing.T) {
	565	+	svc := &Service{
	566	+	inFlight: make(map[string]int),
	567	+	}
	568	+
	569	+	// Add a request and never release it
	570	+	svc.registryInFlight.Add(1)
	571	+
	572	+	drained := svc.DrainRegistryInFlight(30 * time.Millisecond)
	573	+	if drained {
	574	+	t.Fatal("expected DrainRegistryInFlight to return false on timeout, got true")
	575	+	}
	576	+	}


@@ -4,6 +4,7 @@ import (
4	"context"	4		"context"
5	"net/http/httptest"	5		"net/http/httptest"
6	"testing"	6		"testing"
		7	+	"time"
7		8
8	"github.com/bnema/zerowrap"	9		"github.com/bnema/zerowrap"
9	"github.com/stretchr/testify/assert"	10		"github.com/stretchr/testify/assert"

@@ -510,3 +511,66 @@ func TestServeHTTP_MaxBodySize(t *testing.T) {
510	})	511		})
511	}	512		}
512	}	513		}
		514	+
		515	+	func TestRegistryInFlightTracking(t *testing.T) {
		516	+	svc := &Service{
		517	+	inFlight: make(map[string]int),
		518	+	}
		519	+
		520	+	if got := svc.registryInFlight.Load(); got != 0 {
		521	+	t.Fatalf("expected 0 in-flight, got %d", got)
		522	+	}
		523	+
		524	+	svc.registryInFlight.Add(1)
		525	+	if got := svc.registryInFlight.Load(); got != 1 {
		526	+	t.Fatalf("expected 1 in-flight after Add, got %d", got)
		527	+	}
		528	+
		529	+	svc.registryInFlight.Add(-1)
		530	+	if got := svc.registryInFlight.Load(); got != 0 {
		531	+	t.Fatalf("expected 0 in-flight after release, got %d", got)
		532	+	}
		533	+	}
		534	+
		535	+	func TestDrainRegistryInFlight(t *testing.T) {
		536	+	svc := &Service{
		537	+	inFlight: make(map[string]int),
		538	+	}
		539	+
		540	+	svc.registryInFlight.Add(2)
		541	+
		542	+	done := make(chan struct{})
		543	+	go func() {
		544	+	drained := svc.DrainRegistryInFlight(50 * time.Millisecond)
		545	+	if !drained {
		546	+	// Signal failure via done channel by leaving it open — test will time out
		547	+	return
		548	+	}
		549	+	close(done)
		550	+	}()
		551	+
		552	+	time.Sleep(5 * time.Millisecond)
		553	+	svc.registryInFlight.Add(-1)
		554	+	svc.registryInFlight.Add(-1)
		555	+
		556	+	select {
		557	+	case <-done:
		558	+	// good — drained cleanly
		559	+	case <-time.After(500 * time.Millisecond):
		560	+	t.Fatal("DrainRegistryInFlight did not return true after requests completed")
		561	+	}
		562	+	}
		563	+
		564	+	func TestDrainRegistryInFlightTimeout(t *testing.T) {
		565	+	svc := &Service{
		566	+	inFlight: make(map[string]int),
		567	+	}
		568	+
		569	+	// Add a request and never release it
		570	+	svc.registryInFlight.Add(1)
		571	+
		572	+	drained := svc.DrainRegistryInFlight(30 * time.Millisecond)
		573	+	if drained {
		574	+	t.Fatal("expected DrainRegistryInFlight to return false on timeout, got true")
		575	+	}
		576	+	}