fix: activities stream using main context causing app to hang at certain places

Author: kmendell

backend/api/handlers/activities.go

@@ -1,7 +1,10 @@
 package handlers
 
 import (
+	"bufio"
 	"context"
+	"encoding/json"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"testing"
@@ -14,6 +17,7 @@ import (
 	"github.com/getarcaneapp/arcane/backend/v2/internal/database"
 	"github.com/getarcaneapp/arcane/backend/v2/internal/models"
 	"github.com/getarcaneapp/arcane/backend/v2/internal/services"
+	"github.com/getarcaneapp/arcane/types/v2/activity"
 )
 
 func setupActivityHandlerTestDBInternal(t *testing.T) *database.DB {
@@ -97,3 +101,145 @@ func TestActivityHandlerClearHistoryProxiesRemoteEnvironmentInternal(t *testing.
 	require.NoError(t, err)
 	require.EqualValues(t, 7, out.Body.Data.Deleted)
 }
+
+// limitStreamTestDBToSingleConnInternal serializes DB access: the aggregated
+// stream queries from concurrent goroutines, and every extra pooled
+// connection to a :memory: SQLite database is a fresh empty database.
+func limitStreamTestDBToSingleConnInternal(t *testing.T, db *database.DB) {
+	t.Helper()
+	sqlDB, err := db.DB.DB()
+	require.NoError(t, err)
+	sqlDB.SetMaxOpenConns(1)
+}
+
+func createStreamTestRemoteEnvironmentInternal(t *testing.T, db *database.DB, apiURL, token string) {
+	t.Helper()
+	now := time.Now()
+	require.NoError(t, db.Create(&models.Environment{
+		BaseModel: models.BaseModel{
+			ID:        "remote-1",
+			CreatedAt: now,
+			UpdatedAt: &now,
+		},
+		Name:        "Remote",
+		ApiUrl:      apiURL,
+		Status:      string(models.EnvironmentStatusOnline),
+		Enabled:     true,
+		AccessToken: &token,
+	}).Error)
+}
+
+// runStreamAllInternal drives streamAllActivitiesInternal through a pipe and
+// returns each decoded event to onEvent until it reports done or the stream
+// ends; remaining output is drained so a blocked encoder can always finish.
+func runStreamAllInternal(t *testing.T, ctx context.Context, cancel context.CancelFunc, handler *ActivityHandler, onEvent func(activity.StreamEvent) bool) {
+	t.Helper()
+
+	pr, pw := io.Pipe()
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		defer pw.Close()
+		handler.streamAllActivitiesInternal(ctx, 50, json.NewEncoder(pw), func() {})
+	}()
+
+	scanner := bufio.NewScanner(pr)
+	for scanner.Scan() {
+		var event activity.StreamEvent
+		require.NoError(t, json.Unmarshal(scanner.Bytes(), &event))
+		if onEvent(event) {
+			cancel()
+			break
+		}
+	}
+
+	go func() {
+		_, _ = io.Copy(io.Discard, pr)
+	}()
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("stream did not terminate after cancel")
+	}
+}
+
+func TestActivityHandlerStreamAllEmitsEnvironmentScopedEventsInternal(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	db := setupActivityHandlerTestDBInternal(t)
+	limitStreamTestDBToSingleConnInternal(t, db)
+	settingsService, err := services.NewSettingsService(ctx, db)
+	require.NoError(t, err)
+	activityService := services.NewActivityService(db)
+
+	local, err := activityService.StartActivity(ctx, services.StartActivityRequest{EnvironmentID: "0", Type: models.ActivityTypeResourceAction})
+	require.NoError(t, err)
+
+	token := "remote-token"
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write([]byte(`{"success":true,"data":[{"id":"remote-activity-1"}],"pagination":{"totalPages":1,"totalItems":1,"currentPage":1,"itemsPerPage":50}}`))
+	}))
+	defer server.Close()
+	createStreamTestRemoteEnvironmentInternal(t, db, server.URL, token)
+
+	handler := &ActivityHandler{
+		activityService:    activityService,
+		environmentService: services.NewEnvironmentService(db, server.Client(), nil, nil, settingsService, nil),
+	}
+
+	var localSnapshot, remoteSnapshot bool
+	runStreamAllInternal(t, ctx, cancel, handler, func(event activity.StreamEvent) bool {
+		if event.Type == "snapshot" && event.EnvironmentID == "0" && len(event.Activities) == 1 {
+			require.Equal(t, local.ID, event.Activities[0].ID)
+			require.Equal(t, "0", event.Activities[0].SourceEnvironmentID)
+			localSnapshot = true
+		}
+		if event.Type == "snapshot" && event.EnvironmentID == "remote-1" && len(event.Activities) == 1 {
+			require.Equal(t, "remote-activity-1", event.Activities[0].ID)
+			remoteSnapshot = true
+		}
+		return localSnapshot && remoteSnapshot
+	})
+
+	require.True(t, localSnapshot)
+	require.True(t, remoteSnapshot)
+}
+
+func TestActivityHandlerStreamAllRemoteFailureEmitsErrorAndKeepsStreamingInternal(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	db := setupActivityHandlerTestDBInternal(t)
+	limitStreamTestDBToSingleConnInternal(t, db)
+	settingsService, err := services.NewSettingsService(ctx, db)
+	require.NoError(t, err)
+	activityService := services.NewActivityService(db)
+
+	token := "remote-token"
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, "boom", http.StatusInternalServerError)
+	}))
+	defer server.Close()
+	createStreamTestRemoteEnvironmentInternal(t, db, server.URL, token)
+
+	handler := &ActivityHandler{
+		activityService:    activityService,
+		environmentService: services.NewEnvironmentService(db, server.Client(), nil, nil, settingsService, nil),
+	}
+
+	var localSnapshot, remoteError bool
+	runStreamAllInternal(t, ctx, cancel, handler, func(event activity.StreamEvent) bool {
+		if event.Type == "snapshot" && event.EnvironmentID == "0" {
+			localSnapshot = true
+		}
+		if event.Type == "error" && event.EnvironmentID == "remote-1" && event.Error != "" {
+			remoteError = true
+		}
+		return localSnapshot && remoteError
+	})
+
+	require.True(t, localSnapshot)
+	require.True(t, remoteError)
+}

backend/api/handlers/activities_test.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"log/slog"
+	"net"
 	"net/http"
 	"os"
 	"os/signal"
@@ -377,7 +378,14 @@ func runServicesInternal(appCtx context.Context, cfg *config.Config, router http
 	httpHandler, grpcServer := configureTunnelServerInternal(appCtx, cfg, router, tunnelServer, listenAddr)
 	httpHandler, protocols := configureHTTPProtocolsInternal(useTLS, httpHandler)
 
-	srv, err := newHTTPServerInternal(listenAddr, httpHandler, protocols, useTLS, edgeCfg)
+	// Base context for all request contexts. Derived from Background, NOT
+	// appCtx: appCtx carries the app-lifecycle marker value and inheriting it
+	// would make every request context pass utils.IsAppLifecycleContext (see
+	// pkg/projects/cmds.go).
+	baseCtx, cancelBase := context.WithCancel(context.Background())
+	defer cancelBase()
+
+	srv, err := newHTTPServerInternal(baseCtx, listenAddr, httpHandler, protocols, useTLS, edgeCfg) //nolint:contextcheck // baseCtx is deliberately not derived from appCtx, see comment above
 	if err != nil {
 		return err
 	}
@@ -407,6 +415,12 @@ func runServicesInternal(appCtx context.Context, cfg *config.Config, router http
 		slog.InfoContext(appCtx, "Context canceled")
 	}
 
+	// http.Server.Shutdown waits for in-flight handlers but does not cancel
+	// their request contexts; streaming handlers (activity streams, JSON-lines
+	// progress) loop on ctx.Done() and would otherwise pin Shutdown until the
+	// deadline below.
+	cancelBase()
+
 	// Use background context for shutdown as appCtx is already canceled
 	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer shutdownCancel()
@@ -489,7 +503,7 @@ func configureHTTPProtocolsInternal(useTLS bool, handler http.Handler) (http.Han
 	return handler, &protocols
 }
 
-func newHTTPServerInternal(listenAddr string, handler http.Handler, protocols *http.Protocols, useTLS bool, edgeCfg *edge.Config) (*http.Server, error) {
+func newHTTPServerInternal(baseCtx context.Context, listenAddr string, handler http.Handler, protocols *http.Protocols, useTLS bool, edgeCfg *edge.Config) (*http.Server, error) {
 	srv := &http.Server{
 		Addr:              listenAddr,
 		Handler:           handler,
@@ -501,6 +515,10 @@ func newHTTPServerInternal(listenAddr string, handler http.Handler, protocols *h
 		// streaming endpoints (deploy/pull/build progress) need long-lived
 		// connections.
 		IdleTimeout: 120 * time.Second,
+		// BaseContext is canceled right before Shutdown so long-lived
+		// streaming request contexts unblock and graceful shutdown can
+		// complete within its deadline.
+		BaseContext: func(net.Listener) context.Context { return baseCtx },
 	}
 	if !useTLS {
 		return srv, nil

backend/internal/bootstrap/bootstrap.go

@@ -212,6 +212,47 @@ func TestHTTP2APIResponsesDoNotUseAPIGzipInternal(t *testing.T) {
 	}
 }
 
+func TestShutdownCancelsStreamingRequestContextsInternal(t *testing.T) {
+	baseCtx, cancelBase := context.WithCancel(context.Background())
+	defer cancelBase()
+
+	handlerEntered := make(chan struct{})
+	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/x-json-stream")
+		w.WriteHeader(http.StatusOK)
+		if f, ok := w.(http.Flusher); ok {
+			f.Flush()
+		}
+		close(handlerEntered)
+		// Streaming handlers block until their request context ends; Shutdown
+		// alone never cancels it, so this models an open activity stream.
+		<-r.Context().Done()
+	})
+
+	srv, err := newHTTPServerInternal(baseCtx, "127.0.0.1:0", handler, nil, false, nil)
+	require.NoError(t, err)
+
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+
+	errCh := make(chan error, 1)
+	go func() { errCh <- srv.Serve(listener) }()
+
+	resp, err := http.Get("http://" + listener.Addr().String() + "/stream")
+	require.NoError(t, err)
+	defer func() { _ = resp.Body.Close() }()
+	<-handlerEntered
+
+	cancelBase()
+
+	shutdownCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	start := time.Now()
+	require.NoError(t, srv.Shutdown(shutdownCtx))
+	require.Less(t, time.Since(start), time.Second)
+	require.ErrorIs(t, <-errCh, http.ErrServerClosed)
+}
+
 func TestPrepareServerTLSInternal_AgentModeSkipsManagerMTLSValidation(t *testing.T) {
 	cfg := &config.Config{
 		AgentMode:     true,

backend/internal/bootstrap/bootstrap_test.go

@@ -157,12 +157,6 @@ func TestEnvironmentMiddleware_KeepsActivityEndpointsLocal(t *testing.T) {
 			route:  "/environments/:id/activities",
 			path:   "/api/environments/env-edge/activities?limit=50",
 		},
-		{
-			name:   "stream activities",
-			method: http.MethodGet,
-			route:  "/environments/:id/activities/stream",
-			path:   "/api/environments/env-edge/activities/stream?limit=50",
-		},
 	}
 
 	for _, tt := range tests {

backend/internal/middleware/environment_middleware_test.go

@@ -51,7 +51,6 @@ var commandRoutes = []commandRoute{
 	{Method: http.MethodGet, PathPattern: "/api/environments/{id}/image-updates/summary", CommandName: "image_update.summary"},
 
 	{Method: http.MethodGet, PathPattern: "/api/environments/{id}/activities", CommandName: "activity.list"},
-	{Method: http.MethodGet, PathPattern: "/api/environments/{id}/activities/stream", LocalOnly: true},
 	{Method: http.MethodGet, PathPattern: "/api/environments/{id}/activities/{activityId}", CommandName: "activity.inspect"},
 	{Method: http.MethodPost, PathPattern: "/api/environments/{id}/activities/{activityId}/cancel", CommandName: "activity.cancel"},
 	{Method: http.MethodDelete, PathPattern: "/api/environments/{id}/activities/history", CommandName: "activity.history.clear"},

backend/pkg/libarcane/edge/commands.go

@@ -29,7 +29,6 @@ func TestResolveEdgeCommandName(t *testing.T) {
 		{name: "activity inspect", method: "GET", path: "/api/environments/0/activities/activity-1", command: "activity.inspect", shouldHit: true},
 		{name: "activity cancel", method: "POST", path: "/api/environments/0/activities/activity-1/cancel", command: "activity.cancel", shouldHit: true},
 		{name: "activity history clear", method: "DELETE", path: "/api/environments/0/activities/history", command: "activity.history.clear", shouldHit: true},
-		{name: "activity stream remains manager local", method: "GET", path: "/api/environments/0/activities/stream?limit=50", shouldHit: false},
 		{name: "health", method: "HEAD", path: "/api/environments/0/system/health", command: "system.health", shouldHit: true},
 		{name: "swarm node identity", method: "GET", path: "/api/swarm/node-identity", command: "swarm.node_identity", shouldHit: true},
 		{name: "unknown", method: "PATCH", path: "/api/environments/0/containers", shouldHit: false},

backend/pkg/libarcane/edge/commands_test.go

@@ -34,10 +34,14 @@ type AutoHealJob struct {
 	mu       sync.Mutex
 	restarts map[string]*restartRecord
 
-	getDockerClient  func() (*client.Client, error)
-	listContainers   func(ctx context.Context, dockerClient *client.Client) ([]container.Summary, error)
-	inspectContainer func(ctx context.Context, dockerClient *client.Client, containerID string) (container.InspectResponse, error)
-	restartContainer func(ctx context.Context, dockerClient *client.Client, containerID string) error
+	selfIDOnce sync.Once
+	selfID     string
+
+	getDockerClient    func() (*client.Client, error)
+	listContainers     func(ctx context.Context, dockerClient *client.Client) ([]container.Summary, error)
+	inspectContainer   func(ctx context.Context, dockerClient *client.Client, containerID string) (container.InspectResponse, error)
+	restartContainer   func(ctx context.Context, dockerClient *client.Client, containerID string) error
+	getSelfContainerID func() (string, error)
 }
 
 func NewAutoHealJob(
@@ -103,7 +107,8 @@ func (j *AutoHealJob) Run(ctx context.Context) {
 	restartWindowMinutes := j.settingsService.GetIntSetting(ctx, "autoHealRestartWindow", 30)
 	restartWindow := time.Duration(restartWindowMinutes) * time.Minute
 
-	candidates := j.filterCandidatesInternal(containers, excludedContainers)
+	selfID := j.selfContainerIDInternal(ctx)
+	candidates := j.filterCandidatesInternal(containers, excludedContainers, selfID)
 
 	g, groupCtx := errgroup.WithContext(ctx)
 	g.SetLimit(autoHealInspectConcurrency)
@@ -118,9 +123,37 @@ func (j *AutoHealJob) Run(ctx context.Context) {
 	_ = g.Wait()
 }
 
-func (j *AutoHealJob) filterCandidatesInternal(containers []container.Summary, excludedContainers map[string]struct{}) []container.Summary {
+// selfContainerIDInternal resolves and caches the ID (full 64-char or short
+// prefix) of the container Arcane itself runs in. Returns "" when detection
+// fails (e.g. binary running directly on the host), disabling the guard.
+func (j *AutoHealJob) selfContainerIDInternal(ctx context.Context) string {
+	j.selfIDOnce.Do(func() {
+		detect := j.getSelfContainerID
+		if detect == nil {
+			detect = dockerutil.GetCurrentContainerID
+		}
+		id, err := detect()
+		if err != nil {
+			slog.DebugContext(ctx, "auto-heal: could not determine own container ID; self-protection disabled", "error", err)
+			return
+		}
+		j.selfID = strings.ToLower(strings.TrimSpace(id))
+		slog.InfoContext(ctx, "auto-heal: detected own container; it will never be auto-restarted", "container_id", j.selfID)
+	})
+	return j.selfID
+}
+
+func (j *AutoHealJob) filterCandidatesInternal(containers []container.Summary, excludedContainers map[string]struct{}, selfID string) []container.Summary {
 	candidates := make([]container.Summary, 0, len(containers))
 	for _, c := range containers {
+		// Never restart the container Arcane itself runs in: a slow or
+		// mid-startup manager that trips its own healthcheck would otherwise
+		// be restarted by its own auto-heal, in a loop. Prefix match because
+		// hostname-based detection yields the short 12-char ID.
+		if selfID != "" && strings.HasPrefix(strings.ToLower(c.ID), selfID) {
+			continue
+		}
+
 		if libarcane.IsInternalContainer(c.Labels) {
 			continue
 		}

backend/pkg/scheduler/auto_heal_job.go

@@ -22,6 +22,30 @@ func newTestAutoHealJob() *AutoHealJob {
 	}
 }
 
+func TestAutoHeal_FilterCandidates_SkipsSelfContainer(t *testing.T) {
+	job := newTestAutoHealJob()
+
+	selfFullID := "407163929c492b5c4b01a3981f5de4774c37aa8300bd214b4d62412a3dc56468"
+	containers := []container.Summary{
+		{ID: selfFullID, Names: []string{"/arcane"}},
+		{ID: "aaaabbbbccccddddeeeeffff0000111122223333444455556666777788889999", Names: []string{"/other"}},
+	}
+
+	// Hostname-based detection yields the short 12-char ID.
+	candidates := job.filterCandidatesInternal(containers, nil, selfFullID[:12])
+	require.Len(t, candidates, 1)
+	require.Equal(t, "/other", candidates[0].Names[0])
+
+	// cgroup/mountinfo-based detection yields the full ID.
+	candidates = job.filterCandidatesInternal(containers, nil, selfFullID)
+	require.Len(t, candidates, 1)
+	require.Equal(t, "/other", candidates[0].Names[0])
+
+	// Outside Docker no self ID is detected and the guard is a no-op.
+	candidates = job.filterCandidatesInternal(containers, nil, "")
+	require.Len(t, candidates, 2)
+}
+
 func TestAutoHeal_CanRestart_UnderLimit(t *testing.T) {
 	job := newTestAutoHealJob()