fix(container): serialize per-domain deploys to avoid races
- commit
- 6bc6fb8da50469255b8ce4d61c6b789e22c23171
- author
- brice · · 2026-02-05 17:41 UTC
- parent
- 705579a
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
3 files · +112 -1
M internal/adapters/in/cli/remote/client.go +1 -1
| @@ -34,7 +34,7 @@ func NewClient(baseURL string, opts ...ClientOption) *Client { | |||
| 34 | 34 | c := &Client{ | |
| 35 | 35 | baseURL: baseURL, | |
| 36 | 36 | httpClient: &http.Client{ | |
| 37 | - | Timeout: 30 * time.Second, | |
| 37 | + | Timeout: 2 * time.Minute, | |
| 38 | 38 | }, | |
| 39 | 39 | } | |
| 40 | 40 | ||
| @@ -34,7 +34,7 @@ func NewClient(baseURL string, opts ...ClientOption) *Client { | |||||
| 34 | c := &Client{ | 34 | c := &Client{ | ||
| 35 | baseURL: baseURL, | 35 | baseURL: baseURL, | ||
| 36 | httpClient: &http.Client{ | 36 | httpClient: &http.Client{ | ||
| 37 | - | Timeout: 30 * time.Second, | 37 | + | Timeout: 2 * time.Minute, |
| 38 | }, | 38 | }, | ||
| 39 | } | 39 | } | ||
| 40 | 40 | ||||
M internal/usecase/container/service.go +13 -0
| @@ -53,6 +53,7 @@ type Service struct { | |||
| 53 | 53 | containers map[string]*domain.Container | |
| 54 | 54 | attachments map[string][]string // ownerDomain → []containerIDs | |
| 55 | 55 | mu sync.RWMutex | |
| 56 | + | deployMu sync.Map // per-domain deploy mutexes (domain → *sync.Mutex) | |
| 56 | 57 | } | |
| 57 | 58 | ||
| 58 | 59 | // NewService creates a new container service. | |
| @@ -74,9 +75,21 @@ func NewService( | |||
| 74 | 75 | } | |
| 75 | 76 | } | |
| 76 | 77 | ||
| 78 | + | // domainDeployMu returns a per-domain mutex for serializing deploy operations. | |
| 79 | + | func (s *Service) domainDeployMu(domain string) *sync.Mutex { | |
| 80 | + | v, _ := s.deployMu.LoadOrStore(domain, &sync.Mutex{}) | |
| 81 | + | return v.(*sync.Mutex) | |
| 82 | + | } | |
| 83 | + | ||
| 77 | 84 | // Deploy creates and starts a container for the given route. | |
| 78 | 85 | // Implements zero-downtime deployment: new container starts before old one stops. | |
| 79 | 86 | func (s *Service) Deploy(ctx context.Context, route domain.Route) (*domain.Container, error) { | |
| 87 | + | // Serialize deploys for the same domain to prevent race conditions | |
| 88 | + | // (e.g. multiple image.pushed events + explicit deploy call from CLI). | |
| 89 | + | mu := s.domainDeployMu(route.Domain) | |
| 90 | + | mu.Lock() | |
| 91 | + | defer mu.Unlock() | |
| 92 | + | ||
| 80 | 93 | // Enrich context with use case fields for all downstream logs | |
| 81 | 94 | ctx = zerowrap.CtxWithFields(ctx, map[string]any{ | |
| 82 | 95 | zerowrap.FieldLayer: "usecase", | |
| @@ -53,6 +53,7 @@ type Service struct { | |||||
| 53 | containers map[string]*domain.Container | 53 | containers map[string]*domain.Container | ||
| 54 | attachments map[string][]string // ownerDomain → []containerIDs | 54 | attachments map[string][]string // ownerDomain → []containerIDs | ||
| 55 | mu sync.RWMutex | 55 | mu sync.RWMutex | ||
| 56 | + | deployMu sync.Map // per-domain deploy mutexes (domain → *sync.Mutex) | |||
| 56 | } | 57 | } | ||
| 57 | 58 | ||||
| 58 | // NewService creates a new container service. | 59 | // NewService creates a new container service. | ||
| @@ -74,9 +75,21 @@ func NewService( | |||||
| 74 | } | 75 | } | ||
| 75 | } | 76 | } | ||
| 76 | 77 | ||||
| 78 | + | // domainDeployMu returns a per-domain mutex for serializing deploy operations. | |||
| 79 | + | func (s *Service) domainDeployMu(domain string) *sync.Mutex { | |||
| 80 | + | v, _ := s.deployMu.LoadOrStore(domain, &sync.Mutex{}) | |||
| 81 | + | return v.(*sync.Mutex) | |||
| 82 | + | } | |||
| 83 | + | ||||
| 77 | // Deploy creates and starts a container for the given route. | 84 | // Deploy creates and starts a container for the given route. | ||
| 78 | // Implements zero-downtime deployment: new container starts before old one stops. | 85 | // Implements zero-downtime deployment: new container starts before old one stops. | ||
| 79 | func (s *Service) Deploy(ctx context.Context, route domain.Route) (*domain.Container, error) { | 86 | func (s *Service) Deploy(ctx context.Context, route domain.Route) (*domain.Container, error) { | ||
| 87 | + | // Serialize deploys for the same domain to prevent race conditions | |||
| 88 | + | // (e.g. multiple image.pushed events + explicit deploy call from CLI). | |||
| 89 | + | mu := s.domainDeployMu(route.Domain) | |||
| 90 | + | mu.Lock() | |||
| 91 | + | defer mu.Unlock() | |||
| 92 | + | ||||
| 80 | // Enrich context with use case fields for all downstream logs | 93 | // Enrich context with use case fields for all downstream logs | ||
| 81 | ctx = zerowrap.CtxWithFields(ctx, map[string]any{ | 94 | ctx = zerowrap.CtxWithFields(ctx, map[string]any{ | ||
| 82 | zerowrap.FieldLayer: "usecase", | 95 | zerowrap.FieldLayer: "usecase", | ||
M internal/usecase/container/service_test.go +98 -0
| @@ -4,6 +4,7 @@ import ( | |||
| 4 | 4 | "context" | |
| 5 | 5 | "errors" | |
| 6 | 6 | "fmt" | |
| 7 | + | "sync" | |
| 7 | 8 | "testing" | |
| 8 | 9 | "time" | |
| 9 | 10 | ||
| @@ -1417,3 +1418,100 @@ func TestSanitizeName(t *testing.T) { | |||
| 1417 | 1418 | } | |
| 1418 | 1419 | }) | |
| 1419 | 1420 | } | |
| 1421 | + | ||
| 1422 | + | func TestService_Deploy_ConcurrentSameDomain(t *testing.T) { | |
| 1423 | + | // Verify that concurrent Deploy calls for the same domain are serialized | |
| 1424 | + | // and both succeed without container name conflicts. | |
| 1425 | + | runtime := mocks.NewMockContainerRuntime(t) | |
| 1426 | + | envLoader := mocks.NewMockEnvLoader(t) | |
| 1427 | + | eventBus := mocks.NewMockEventPublisher(t) | |
| 1428 | + | ||
| 1429 | + | config := Config{ | |
| 1430 | + | ReadinessDelay: time.Millisecond, | |
| 1431 | + | } | |
| 1432 | + | svc := NewService(runtime, envLoader, eventBus, nil, config) | |
| 1433 | + | ctx := testContext() | |
| 1434 | + | ||
| 1435 | + | route := domain.Route{ | |
| 1436 | + | Domain: "test.example.com", | |
| 1437 | + | Image: "myapp:latest", | |
| 1438 | + | } | |
| 1439 | + | ||
| 1440 | + | // Track call order to verify serialization | |
| 1441 | + | var callOrder []string | |
| 1442 | + | var callMu sync.Mutex | |
| 1443 | + | ||
| 1444 | + | // Setup mocks that will be called by both deploys sequentially. | |
| 1445 | + | // First deploy: no existing container → creates gordon-test.example.com | |
| 1446 | + | // Second deploy: sees first container → creates gordon-test.example.com-new | |
| 1447 | + | ||
| 1448 | + | runtime.EXPECT().ListContainers(mock.Anything, true).Return([]*domain.Container{}, nil).Times(2) | |
| 1449 | + | runtime.EXPECT().ListImages(mock.Anything).Return([]string{"myapp:latest"}, nil).Times(2) | |
| 1450 | + | runtime.EXPECT().GetImageExposedPorts(mock.Anything, "myapp:latest").Return([]int{8080}, nil).Times(2) | |
| 1451 | + | envLoader.EXPECT().LoadEnv(mock.Anything, "test.example.com").Return(nil, nil).Times(2) | |
| 1452 | + | runtime.EXPECT().InspectImageEnv(mock.Anything, "myapp:latest").Return(nil, nil).Times(2) | |
| 1453 | + | ||
| 1454 | + | createCall := 0 | |
| 1455 | + | runtime.EXPECT().CreateContainer(mock.Anything, mock.AnythingOfType("*domain.ContainerConfig")). | |
| 1456 | + | RunAndReturn(func(_ context.Context, cfg *domain.ContainerConfig) (*domain.Container, error) { | |
| 1457 | + | callMu.Lock() | |
| 1458 | + | createCall++ | |
| 1459 | + | n := createCall | |
| 1460 | + | callOrder = append(callOrder, fmt.Sprintf("create-%d:%s", n, cfg.Name)) | |
| 1461 | + | callMu.Unlock() | |
| 1462 | + | return &domain.Container{ | |
| 1463 | + | ID: fmt.Sprintf("container-%d", n), | |
| 1464 | + | Name: cfg.Name, | |
| 1465 | + | Image: cfg.Image, | |
| 1466 | + | Status: "created", | |
| 1467 | + | }, nil | |
| 1468 | + | }) | |
| 1469 | + | ||
| 1470 | + | runtime.EXPECT().StartContainer(mock.Anything, mock.AnythingOfType("string")).Return(nil).Times(2) | |
| 1471 | + | // IsContainerRunning is called 2x per deploy in waitForReady (initial check + after delay) | |
| 1472 | + | runtime.EXPECT().IsContainerRunning(mock.Anything, mock.AnythingOfType("string")).Return(true, nil).Times(4) | |
| 1473 | + | ||
| 1474 | + | runtime.EXPECT().InspectContainer(mock.Anything, mock.AnythingOfType("string")). | |
| 1475 | + | RunAndReturn(func(_ context.Context, id string) (*domain.Container, error) { | |
| 1476 | + | return &domain.Container{ | |
| 1477 | + | ID: id, | |
| 1478 | + | Name: "gordon-test.example.com", | |
| 1479 | + | Image: "myapp:latest", | |
| 1480 | + | Status: "running", | |
| 1481 | + | Ports: []int{8080}, | |
| 1482 | + | }, nil | |
| 1483 | + | }) | |
| 1484 | + | ||
| 1485 | + | eventBus.EXPECT().Publish(domain.EventContainerDeployed, mock.Anything).Return(nil).Times(2) | |
| 1486 | + | ||
| 1487 | + | // The second deploy will see the first container and do zero-downtime swap, | |
| 1488 | + | // which means it will also stop+remove+rename the old container. | |
| 1489 | + | runtime.EXPECT().StopContainer(mock.Anything, mock.AnythingOfType("string")).Return(nil).Maybe() | |
| 1490 | + | runtime.EXPECT().RemoveContainer(mock.Anything, mock.AnythingOfType("string"), true).Return(nil).Maybe() | |
| 1491 | + | runtime.EXPECT().RenameContainer(mock.Anything, mock.AnythingOfType("string"), mock.AnythingOfType("string")).Return(nil).Maybe() | |
| 1492 | + | ||
| 1493 | + | // Launch two concurrent deploys | |
| 1494 | + | var wg sync.WaitGroup | |
| 1495 | + | errs := make([]error, 2) | |
| 1496 | + | ||
| 1497 | + | wg.Add(2) | |
| 1498 | + | for i := range 2 { | |
| 1499 | + | go func(idx int) { | |
| 1500 | + | defer wg.Done() | |
| 1501 | + | _, errs[idx] = svc.Deploy(ctx, route) | |
| 1502 | + | }(i) | |
| 1503 | + | } | |
| 1504 | + | wg.Wait() | |
| 1505 | + | ||
| 1506 | + | // Both should succeed (second deploy waits for the first to finish) | |
| 1507 | + | assert.NoError(t, errs[0], "first deploy should succeed") | |
| 1508 | + | assert.NoError(t, errs[1], "second deploy should succeed") | |
| 1509 | + | ||
| 1510 | + | // Verify creates were serialized (not interleaved) | |
| 1511 | + | callMu.Lock() | |
| 1512 | + | assert.Len(t, callOrder, 2, "should have exactly 2 create calls") | |
| 1513 | + | // First deploy uses canonical name, second uses -new suffix (zero-downtime) | |
| 1514 | + | assert.Contains(t, callOrder[0], "gordon-test.example.com", "first create should use canonical name") | |
| 1515 | + | assert.Contains(t, callOrder[1], "gordon-test.example.com-new", "second create should use -new suffix") | |
| 1516 | + | callMu.Unlock() | |
| 1517 | + | } | |
| @@ -4,6 +4,7 @@ import ( | |||||
| 4 | "context" | 4 | "context" | ||
| 5 | "errors" | 5 | "errors" | ||
| 6 | "fmt" | 6 | "fmt" | ||
| 7 | + | "sync" | |||
| 7 | "testing" | 8 | "testing" | ||
| 8 | "time" | 9 | "time" | ||
| 9 | 10 | ||||
| @@ -1417,3 +1418,100 @@ func TestSanitizeName(t *testing.T) { | |||||
| 1417 | } | 1418 | } | ||
| 1418 | }) | 1419 | }) | ||
| 1419 | } | 1420 | } | ||
| 1421 | + | ||||
| 1422 | + | func TestService_Deploy_ConcurrentSameDomain(t *testing.T) { | |||
| 1423 | + | // Verify that concurrent Deploy calls for the same domain are serialized | |||
| 1424 | + | // and both succeed without container name conflicts. | |||
| 1425 | + | runtime := mocks.NewMockContainerRuntime(t) | |||
| 1426 | + | envLoader := mocks.NewMockEnvLoader(t) | |||
| 1427 | + | eventBus := mocks.NewMockEventPublisher(t) | |||
| 1428 | + | ||||
| 1429 | + | config := Config{ | |||
| 1430 | + | ReadinessDelay: time.Millisecond, | |||
| 1431 | + | } | |||
| 1432 | + | svc := NewService(runtime, envLoader, eventBus, nil, config) | |||
| 1433 | + | ctx := testContext() | |||
| 1434 | + | ||||
| 1435 | + | route := domain.Route{ | |||
| 1436 | + | Domain: "test.example.com", | |||
| 1437 | + | Image: "myapp:latest", | |||
| 1438 | + | } | |||
| 1439 | + | ||||
| 1440 | + | // Track call order to verify serialization | |||
| 1441 | + | var callOrder []string | |||
| 1442 | + | var callMu sync.Mutex | |||
| 1443 | + | ||||
| 1444 | + | // Setup mocks that will be called by both deploys sequentially. | |||
| 1445 | + | // First deploy: no existing container → creates gordon-test.example.com | |||
| 1446 | + | // Second deploy: sees first container → creates gordon-test.example.com-new | |||
| 1447 | + | ||||
| 1448 | + | runtime.EXPECT().ListContainers(mock.Anything, true).Return([]*domain.Container{}, nil).Times(2) | |||
| 1449 | + | runtime.EXPECT().ListImages(mock.Anything).Return([]string{"myapp:latest"}, nil).Times(2) | |||
| 1450 | + | runtime.EXPECT().GetImageExposedPorts(mock.Anything, "myapp:latest").Return([]int{8080}, nil).Times(2) | |||
| 1451 | + | envLoader.EXPECT().LoadEnv(mock.Anything, "test.example.com").Return(nil, nil).Times(2) | |||
| 1452 | + | runtime.EXPECT().InspectImageEnv(mock.Anything, "myapp:latest").Return(nil, nil).Times(2) | |||
| 1453 | + | ||||
| 1454 | + | createCall := 0 | |||
| 1455 | + | runtime.EXPECT().CreateContainer(mock.Anything, mock.AnythingOfType("*domain.ContainerConfig")). | |||
| 1456 | + | RunAndReturn(func(_ context.Context, cfg *domain.ContainerConfig) (*domain.Container, error) { | |||
| 1457 | + | callMu.Lock() | |||
| 1458 | + | createCall++ | |||
| 1459 | + | n := createCall | |||
| 1460 | + | callOrder = append(callOrder, fmt.Sprintf("create-%d:%s", n, cfg.Name)) | |||
| 1461 | + | callMu.Unlock() | |||
| 1462 | + | return &domain.Container{ | |||
| 1463 | + | ID: fmt.Sprintf("container-%d", n), | |||
| 1464 | + | Name: cfg.Name, | |||
| 1465 | + | Image: cfg.Image, | |||
| 1466 | + | Status: "created", | |||
| 1467 | + | }, nil | |||
| 1468 | + | }) | |||
| 1469 | + | ||||
| 1470 | + | runtime.EXPECT().StartContainer(mock.Anything, mock.AnythingOfType("string")).Return(nil).Times(2) | |||
| 1471 | + | // IsContainerRunning is called 2x per deploy in waitForReady (initial check + after delay) | |||
| 1472 | + | runtime.EXPECT().IsContainerRunning(mock.Anything, mock.AnythingOfType("string")).Return(true, nil).Times(4) | |||
| 1473 | + | ||||
| 1474 | + | runtime.EXPECT().InspectContainer(mock.Anything, mock.AnythingOfType("string")). | |||
| 1475 | + | RunAndReturn(func(_ context.Context, id string) (*domain.Container, error) { | |||
| 1476 | + | return &domain.Container{ | |||
| 1477 | + | ID: id, | |||
| 1478 | + | Name: "gordon-test.example.com", | |||
| 1479 | + | Image: "myapp:latest", | |||
| 1480 | + | Status: "running", | |||
| 1481 | + | Ports: []int{8080}, | |||
| 1482 | + | }, nil | |||
| 1483 | + | }) | |||
| 1484 | + | ||||
| 1485 | + | eventBus.EXPECT().Publish(domain.EventContainerDeployed, mock.Anything).Return(nil).Times(2) | |||
| 1486 | + | ||||
| 1487 | + | // The second deploy will see the first container and do zero-downtime swap, | |||
| 1488 | + | // which means it will also stop+remove+rename the old container. | |||
| 1489 | + | runtime.EXPECT().StopContainer(mock.Anything, mock.AnythingOfType("string")).Return(nil).Maybe() | |||
| 1490 | + | runtime.EXPECT().RemoveContainer(mock.Anything, mock.AnythingOfType("string"), true).Return(nil).Maybe() | |||
| 1491 | + | runtime.EXPECT().RenameContainer(mock.Anything, mock.AnythingOfType("string"), mock.AnythingOfType("string")).Return(nil).Maybe() | |||
| 1492 | + | ||||
| 1493 | + | // Launch two concurrent deploys | |||
| 1494 | + | var wg sync.WaitGroup | |||
| 1495 | + | errs := make([]error, 2) | |||
| 1496 | + | ||||
| 1497 | + | wg.Add(2) | |||
| 1498 | + | for i := range 2 { | |||
| 1499 | + | go func(idx int) { | |||
| 1500 | + | defer wg.Done() | |||
| 1501 | + | _, errs[idx] = svc.Deploy(ctx, route) | |||
| 1502 | + | }(i) | |||
| 1503 | + | } | |||
| 1504 | + | wg.Wait() | |||
| 1505 | + | ||||
| 1506 | + | // Both should succeed (second deploy waits for the first to finish) | |||
| 1507 | + | assert.NoError(t, errs[0], "first deploy should succeed") | |||
| 1508 | + | assert.NoError(t, errs[1], "second deploy should succeed") | |||
| 1509 | + | ||||
| 1510 | + | // Verify creates were serialized (not interleaved) | |||
| 1511 | + | callMu.Lock() | |||
| 1512 | + | assert.Len(t, callOrder, 2, "should have exactly 2 create calls") | |||
| 1513 | + | // First deploy uses canonical name, second uses -new suffix (zero-downtime) | |||
| 1514 | + | assert.Contains(t, callOrder[0], "gordon-test.example.com", "first create should use canonical name") | |||
| 1515 | + | assert.Contains(t, callOrder[1], "gordon-test.example.com-new", "second create should use -new suffix") | |||
| 1516 | + | callMu.Unlock() | |||
| 1517 | + | } | |||