fix(concurrency): fix lock scoping in container service and event bus

Move HealthCheck and SyncContainers lock acquisition to method-level scoping
to prevent holding locks unnecessarily during long-running operations.
Improve event bus error handling and ensure attachment cleanup occurs reliably.

commit: 436305710ec07fee08496f7ed12f4c48b9d0f5d3
author: brice · 8mo ago · 2026-02-06 05:32 UTC
parent: 02b51ba
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +85 -36

M internal/adapters/out/eventbus/inmemory.go +30 -24


@@ -74,7 +74,15 @@ func (bus *InMemory) Publish(eventType domain.EventType, payload any) error {
74	74		return nil
75	75		case <-bus.ctx.Done():
76	76		return fmt.Errorf("event bus is stopped")
77		-	default:
	77	+	case <-time.After(5 * time.Second):
	78	+	bus.log.Error().
	79	+	Str(zerowrap.FieldLayer, "adapter").
	80	+	Str(zerowrap.FieldAdapter, "eventbus").
	81	+	Str("event_id", event.ID).
	82	+	Str(zerowrap.FieldEvent, string(event.Type)).
	83	+	Str("image_name", event.ImageName).
	84	+	Str("tag", event.Tag).
	85	+	Msg("event channel is full, dropping event after 5s timeout")
78	86		return fmt.Errorf("event channel is full, dropping event %s", event.ID)
79	87		}
80	88		}

@@ -178,29 +186,27 @@ func (bus *InMemory) handleEvent(event domain.Event) {
178	186
179	187		for _, handler := range handlers {
180	188		if handler.CanHandle(event.Type) {
181		-	go func(h out.EventHandler) {
182		-	start := time.Now()
183		-
184		-	if err := h.Handle(event); err != nil {
185		-	bus.log.Error().
186		-	Str(zerowrap.FieldLayer, "adapter").
187		-	Str(zerowrap.FieldAdapter, "eventbus").
188		-	Err(err).
189		-	Str("event_id", event.ID).
190		-	Str(zerowrap.FieldEvent, string(event.Type)).
191		-	Str(zerowrap.FieldHandler, fmt.Sprintf("%T", h)).
192		-	Msg("error handling event")
193		-	} else {
194		-	bus.log.Debug().
195		-	Str(zerowrap.FieldLayer, "adapter").
196		-	Str(zerowrap.FieldAdapter, "eventbus").
197		-	Str("event_id", event.ID).
198		-	Str(zerowrap.FieldEvent, string(event.Type)).
199		-	Str(zerowrap.FieldHandler, fmt.Sprintf("%T", h)).
200		-	Dur(zerowrap.FieldDuration, time.Since(start)).
201		-	Msg("event handled successfully")
202		-	}
203		-	}(handler)
	189	+	start := time.Now()
	190	+
	191	+	if err := handler.Handle(event); err != nil {
	192	+	bus.log.Error().
	193	+	Str(zerowrap.FieldLayer, "adapter").
	194	+	Str(zerowrap.FieldAdapter, "eventbus").
	195	+	Err(err).
	196	+	Str("event_id", event.ID).
	197	+	Str(zerowrap.FieldEvent, string(event.Type)).
	198	+	Str(zerowrap.FieldHandler, fmt.Sprintf("%T", handler)).
	199	+	Msg("error handling event")
	200	+	} else {
	201	+	bus.log.Debug().
	202	+	Str(zerowrap.FieldLayer, "adapter").
	203	+	Str(zerowrap.FieldAdapter, "eventbus").
	204	+	Str("event_id", event.ID).
	205	+	Str(zerowrap.FieldEvent, string(event.Type)).
	206	+	Str(zerowrap.FieldHandler, fmt.Sprintf("%T", handler)).
	207	+	Dur(zerowrap.FieldDuration, time.Since(start)).
	208	+	Msg("event handled successfully")
	209	+	}
204	210		}
205	211		}
206	212		}


@@ -74,7 +74,15 @@ func (bus *InMemory) Publish(eventType domain.EventType, payload any) error {
74		return nil	74		return nil
75		case <-bus.ctx.Done():	75		case <-bus.ctx.Done():
76		return fmt.Errorf("event bus is stopped")	76		return fmt.Errorf("event bus is stopped")
77	-	default:	77	+	case <-time.After(5 * time.Second):
			78	+	bus.log.Error().
			79	+	Str(zerowrap.FieldLayer, "adapter").
			80	+	Str(zerowrap.FieldAdapter, "eventbus").
			81	+	Str("event_id", event.ID).
			82	+	Str(zerowrap.FieldEvent, string(event.Type)).
			83	+	Str("image_name", event.ImageName).
			84	+	Str("tag", event.Tag).
			85	+	Msg("event channel is full, dropping event after 5s timeout")
78		return fmt.Errorf("event channel is full, dropping event %s", event.ID)	86		return fmt.Errorf("event channel is full, dropping event %s", event.ID)
79		}	87		}
80		}	88		}

@@ -178,29 +186,27 @@ func (bus *InMemory) handleEvent(event domain.Event) {
178			186
179		for _, handler := range handlers {	187		for _, handler := range handlers {
180		if handler.CanHandle(event.Type) {	188		if handler.CanHandle(event.Type) {
181	-	go func(h out.EventHandler) {	189	+	start := time.Now()
182	-	start := time.Now()	190	+
183	-		191	+	if err := handler.Handle(event); err != nil {
184	-	if err := h.Handle(event); err != nil {	192	+	bus.log.Error().
185	-	bus.log.Error().	193	+	Str(zerowrap.FieldLayer, "adapter").
186	-	Str(zerowrap.FieldLayer, "adapter").	194	+	Str(zerowrap.FieldAdapter, "eventbus").
187	-	Str(zerowrap.FieldAdapter, "eventbus").	195	+	Err(err).
188	-	Err(err).	196	+	Str("event_id", event.ID).
189	-	Str("event_id", event.ID).	197	+	Str(zerowrap.FieldEvent, string(event.Type)).
190	-	Str(zerowrap.FieldEvent, string(event.Type)).	198	+	Str(zerowrap.FieldHandler, fmt.Sprintf("%T", handler)).
191	-	Str(zerowrap.FieldHandler, fmt.Sprintf("%T", h)).	199	+	Msg("error handling event")
192	-	Msg("error handling event")	200	+	} else {
193	-	} else {	201	+	bus.log.Debug().
194	-	bus.log.Debug().	202	+	Str(zerowrap.FieldLayer, "adapter").
195	-	Str(zerowrap.FieldLayer, "adapter").	203	+	Str(zerowrap.FieldAdapter, "eventbus").
196	-	Str(zerowrap.FieldAdapter, "eventbus").	204	+	Str("event_id", event.ID).
197	-	Str("event_id", event.ID).	205	+	Str(zerowrap.FieldEvent, string(event.Type)).
198	-	Str(zerowrap.FieldEvent, string(event.Type)).	206	+	Str(zerowrap.FieldHandler, fmt.Sprintf("%T", handler)).
199	-	Str(zerowrap.FieldHandler, fmt.Sprintf("%T", h)).	207	+	Dur(zerowrap.FieldDuration, time.Since(start)).
200	-	Dur(zerowrap.FieldDuration, time.Since(start)).	208	+	Msg("event handled successfully")
201	-	Msg("event handled successfully")	209	+	}
202	-	}
203	-	}(handler)
204		}	210		}
205		}	211		}
206		}	212		}

M internal/usecase/container/service.go +55 -12


@@ -213,7 +213,7 @@ func (s Service) Deploy(ctx context.Context, route domain.Route) (domain.Conta
213	213
214	214		// Deploy attachments BEFORE main container (dependencies first)
215	215		if err := s.deployAttachments(ctx, route.Domain, networkName); err != nil {
216		-	log.WrapErr(err, "failed to deploy some attachments")
	216	+	return nil, log.WrapErr(err, "failed to deploy attachments")
217	217		}
218	218
219	219		// Get exposed ports (use actual image ref from pull)

@@ -371,8 +371,9 @@ func (s *Service) Remove(ctx context.Context, containerID string, force bool) er
371	371		}
372	372		}
373	373
374		-	// Find domain for cleanup
	374	+	// Find domain and attachment IDs for cleanup
375	375		var containerDomain string
	376	+	var attachmentIDs []string
376	377		s.mu.RLock()
377	378		for d, c := range s.containers {
378	379		if c.ID == containerID {

@@ -380,6 +381,9 @@ func (s *Service) Remove(ctx context.Context, containerID string, force bool) er
380	381		break
381	382		}
382	383		}
	384	+	if containerDomain != "" {
	385	+	attachmentIDs = append(attachmentIDs, s.attachments[containerDomain]...)
	386	+	}
383	387		s.mu.RUnlock()
384	388
385	389		if err := s.runtime.RemoveContainer(ctx, containerID, force); err != nil {

@@ -393,6 +397,21 @@ func (s *Service) Remove(ctx context.Context, containerID string, force bool) er
393	397		}
394	398		}
395	399
	400	+	// Clean up attachment containers
	401	+	for _, attachID := range attachmentIDs {
	402	+	if s.logWriter != nil {
	403	+	if err := s.logWriter.StopLogging(attachID); err != nil {
	404	+	log.Warn().Err(err).Str("attachment_id", attachID).Msg("failed to stop attachment log collection")
	405	+	}
	406	+	}
	407	+	if err := s.runtime.StopContainer(ctx, attachID); err != nil {
	408	+	log.Warn().Err(err).Str("attachment_id", attachID).Msg("failed to stop attachment container")
	409	+	}
	410	+	if err := s.runtime.RemoveContainer(ctx, attachID, true); err != nil {
	411	+	log.Warn().Err(err).Str("attachment_id", attachID).Msg("failed to remove attachment container")
	412	+	}
	413	+	}
	414	+
396	415		// Remove from tracking
397	416		s.mu.Lock()
398	417		var removedDomain string

@@ -404,6 +423,9 @@ func (s *Service) Remove(ctx context.Context, containerID string, force bool) er
404	423		break
405	424		}
406	425		}
	426	+	if removedDomain != "" {
	427	+	delete(s.attachments, removedDomain)
	428	+	}
407	429		s.mu.Unlock()
408	430
409	431		// Note: We intentionally do NOT clean up deployMu entries to avoid a race condition:

@@ -532,14 +554,19 @@ func (s *Service) HealthCheck(ctx context.Context) map[string]bool {
532	554		})
533	555		log := zerowrap.FromCtx(ctx)
534	556
	557	+	// Copy container IDs under lock, then release before making Docker API calls
535	558		s.mu.RLock()
536		-	defer s.mu.RUnlock()
537		-
538		-	health := make(map[string]bool, len(s.containers))
	559	+	snapshot := make(map[string]string, len(s.containers))
539	560		for d, c := range s.containers {
540		-	running, err := s.runtime.IsContainerRunning(ctx, c.ID)
	561	+	snapshot[d] = c.ID
	562	+	}
	563	+	s.mu.RUnlock()
	564	+
	565	+	health := make(map[string]bool, len(snapshot))
	566	+	for d, id := range snapshot {
	567	+	running, err := s.runtime.IsContainerRunning(ctx, id)
541	568		if err != nil {
542		-	log.WrapErrWithFields(err, "health check failed", map[string]any{"domain": d, zerowrap.FieldEntityID: c.ID})
	569	+	log.WrapErrWithFields(err, "health check failed", map[string]any{"domain": d, zerowrap.FieldEntityID: id})
543	570		health[d] = false
544	571		} else {
545	572		health[d] = running

@@ -556,9 +583,7 @@ func (s *Service) SyncContainers(ctx context.Context) error {
556	583		})
557	584		log := zerowrap.FromCtx(ctx)
558	585
559		-	s.mu.Lock()
560		-	defer s.mu.Unlock()
561		-
	586	+	// List containers without holding lock to avoid blocking during Docker API call
562	587		allContainers, err := s.runtime.ListContainers(ctx, false)
563	588		if err != nil {
564	589		return log.WrapErr(err, "failed to list containers")

@@ -573,7 +598,10 @@ func (s *Service) SyncContainers(ctx context.Context) error {
573	598		}
574	599		}
575	600
	601	+	s.mu.Lock()
576	602		s.containers = managed
	603	+	s.mu.Unlock()
	604	+
577	605		log.Info().Int(zerowrap.FieldCount, len(managed)).Msg("container state synchronized")
578	606		return nil
579	607		}

@@ -649,7 +677,9 @@ func (s *Service) Shutdown(ctx context.Context) error {
649	677
650	678		// UpdateConfig updates the service configuration.
651	679		func (s *Service) UpdateConfig(config Config) {
	680	+	s.mu.Lock()
652	681		s.config = config
	682	+	s.mu.Unlock()
653	683		}
654	684
655	685		// Helper methods

@@ -1138,12 +1168,17 @@ func (s *Service) deployAttachments(ctx context.Context, domainName, networkName
1138	1168		}
1139	1169
1140	1170		log := zerowrap.FromCtx(ctx)
	1171	+	var failed int
1141	1172		for _, svc := range attachments {
1142	1173		if err := s.deployAttachedService(ctx, domainName, svc, networkName); err != nil {
1143	1174		log.WrapErrWithFields(err, "failed to deploy attachment", map[string]any{zerowrap.FieldService: svc, "domain": domainName})
	1175	+	failed++
1144	1176		}
1145	1177		}
1146	1178
	1179	+	if failed > 0 {
	1180	+	return fmt.Errorf("failed to deploy %d of %d attachment(s)", failed, len(attachments))
	1181	+	}
1147	1182		return nil
1148	1183		}
1149	1184

@@ -1557,7 +1592,11 @@ func (s *Service) waitForReady(ctx context.Context, containerID string) error {
1557	1592		if i == 29 {
1558	1593		return fmt.Errorf("container did not start within 30 seconds")
1559	1594		}
1560		-	time.Sleep(time.Second)
	1595	+	select {
	1596	+	case <-time.After(time.Second):
	1597	+	case <-ctx.Done():
	1598	+	return ctx.Err()
	1599	+	}
1561	1600		}
1562	1601
1563	1602		// Additional readiness delay (configurable, default 5 seconds)

@@ -1567,7 +1606,11 @@ func (s *Service) waitForReady(ctx context.Context, containerID string) error {
1567	1606		}
1568	1607
1569	1608		log.Debug().Dur("delay", delay).Msg("waiting for container readiness")
1570		-	time.Sleep(delay)
	1609	+	select {
	1610	+	case <-time.After(delay):
	1611	+	case <-ctx.Done():
	1612	+	return ctx.Err()
	1613	+	}
1571	1614
1572	1615		// Verify still running after delay
1573	1616		running, err := s.runtime.IsContainerRunning(ctx, containerID)

 	// Deploy attachments BEFORE main container (dependencies first)
 	if err := s.deployAttachments(ctx, route.Domain, networkName); err != nil {
-		log.WrapErr(err, "failed to deploy some attachments")
+		return nil, log.WrapErr(err, "failed to deploy attachments")
 	}
 	// Get exposed ports (use actual image ref from pull)
 		}
 	}
-	// Find domain for cleanup
+	// Find domain and attachment IDs for cleanup
 	var containerDomain string
+	var attachmentIDs []string
 	s.mu.RLock()
 	for d, c := range s.containers {
 		if c.ID == containerID {
 			break
 		}
 	}
+	if containerDomain != "" {
+		attachmentIDs = append(attachmentIDs, s.attachments[containerDomain]...)
+	}
 	s.mu.RUnlock()
 	if err := s.runtime.RemoveContainer(ctx, containerID, force); err != nil {
 		}
 	}
+	// Clean up attachment containers
+	for _, attachID := range attachmentIDs {
+		if s.logWriter != nil {
+			if err := s.logWriter.StopLogging(attachID); err != nil {
+				log.Warn().Err(err).Str("attachment_id", attachID).Msg("failed to stop attachment log collection")
+			}
+		}
+		if err := s.runtime.StopContainer(ctx, attachID); err != nil {
+			log.Warn().Err(err).Str("attachment_id", attachID).Msg("failed to stop attachment container")
+		}
+		if err := s.runtime.RemoveContainer(ctx, attachID, true); err != nil {
+			log.Warn().Err(err).Str("attachment_id", attachID).Msg("failed to remove attachment container")
+		}
+	}
 	// Remove from tracking
 	s.mu.Lock()
 	var removedDomain string
 			break
 		}
 	}
+	if removedDomain != "" {
+		delete(s.attachments, removedDomain)
+	}
 	s.mu.Unlock()
 	// Note: We intentionally do NOT clean up deployMu entries to avoid a race condition:
 	})
 	log := zerowrap.FromCtx(ctx)
+	// Copy container IDs under lock, then release before making Docker API calls
 	s.mu.RLock()
-	defer s.mu.RUnlock()
+	snapshot := make(map[string]string, len(s.containers))
-	health := make(map[string]bool, len(s.containers))
 	for d, c := range s.containers {
-		running, err := s.runtime.IsContainerRunning(ctx, c.ID)
+		snapshot[d] = c.ID
+	}
+	s.mu.RUnlock()
+	health := make(map[string]bool, len(snapshot))
+	for d, id := range snapshot {
+		running, err := s.runtime.IsContainerRunning(ctx, id)
 		if err != nil {
-			log.WrapErrWithFields(err, "health check failed", map[string]any{"domain": d, zerowrap.FieldEntityID: c.ID})
+			log.WrapErrWithFields(err, "health check failed", map[string]any{"domain": d, zerowrap.FieldEntityID: id})
 			health[d] = false
 		} else {
 			health[d] = running
 	})
 	log := zerowrap.FromCtx(ctx)
-	s.mu.Lock()
+	// List containers without holding lock to avoid blocking during Docker API call
-	defer s.mu.Unlock()
 	allContainers, err := s.runtime.ListContainers(ctx, false)
 	if err != nil {
 		return log.WrapErr(err, "failed to list containers")
 		}
 	}
+	s.mu.Lock()
 	s.containers = managed
+	s.mu.Unlock()
 	log.Info().Int(zerowrap.FieldCount, len(managed)).Msg("container state synchronized")
 	return nil
 }
 // UpdateConfig updates the service configuration.
 func (s *Service) UpdateConfig(config Config) {
+	s.mu.Lock()
 	s.config = config
+	s.mu.Unlock()
 }
 // Helper methods
 	}
 	log := zerowrap.FromCtx(ctx)
+	var failed int
 	for _, svc := range attachments {
 		if err := s.deployAttachedService(ctx, domainName, svc, networkName); err != nil {
 			log.WrapErrWithFields(err, "failed to deploy attachment", map[string]any{zerowrap.FieldService: svc, "domain": domainName})
+			failed++
 		}
 	}
+	if failed > 0 {
+		return fmt.Errorf("failed to deploy %d of %d attachment(s)", failed, len(attachments))
+	}
 	return nil
 }
 		if i == 29 {
 			return fmt.Errorf("container did not start within 30 seconds")
 		}
-		time.Sleep(time.Second)
+		select {
+		case <-time.After(time.Second):
+		case <-ctx.Done():
+			return ctx.Err()
+		}
 	}
 	// Additional readiness delay (configurable, default 5 seconds)
 	}
 	log.Debug().Dur("delay", delay).Msg("waiting for container readiness")
-	time.Sleep(delay)
+	select {
+	case <-time.After(delay):
+	case <-ctx.Done():
+		return ctx.Err()
+	}
 	// Verify still running after delay
 	running, err := s.runtime.IsContainerRunning(ctx, containerID)