feat(registry): add deploy event suppression for CLI-managed pushes

commit: 82c3ab44d35baed671b6cc2d67329277f9ca2f04
author: brice · 7mo ago · 2026-03-04 06:38 UTC
parent: 40bf8ea
signature: unsigned

2 files · +82 -11

M internal/usecase/registry/service.go +43 -11


@@ -7,6 +7,8 @@ import (
7	7		"fmt"
8	8		"io"
9	9		"strings"
	10	+	"sync"
	11	+	"time"
10	12
11	13		"github.com/bnema/zerowrap"
12	14		"go.opentelemetry.io/otel"

@@ -24,10 +26,11 @@ var registryTracer = otel.Tracer("gordon.registry")
24	26
25	27		// Service implements the RegistryService interface.
26	28		type Service struct {
27		-	blobStorage out.BlobStorage
28		-	manifestStorage out.ManifestStorage
29		-	eventBus out.EventPublisher
30		-	metrics *telemetry.Metrics
	29	+	blobStorage out.BlobStorage
	30	+	manifestStorage out.ManifestStorage
	31	+	eventBus out.EventPublisher
	32	+	metrics *telemetry.Metrics
	33	+	suppressedImages sync.Map // imageName -> *time.Timer
31	34		}
32	35
33	36		// SetMetrics sets the telemetry metrics for the registry service.

@@ -48,6 +51,31 @@ func NewService(
48	51		}
49	52		}
50	53
	54	+	// SuppressDeployEvent marks an image name to skip image.pushed events.
	55	+	// The suppression auto-expires after 2 minutes to prevent leaks.
	56	+	func (s *Service) SuppressDeployEvent(imageName string) {
	57	+	timer := time.AfterFunc(2*time.Minute, func() {
	58	+	s.suppressedImages.Delete(imageName)
	59	+	})
	60	+	if existing, loaded := s.suppressedImages.LoadOrStore(imageName, timer); loaded {
	61	+	existing.(*time.Timer).Stop()
	62	+	s.suppressedImages.Store(imageName, timer)
	63	+	}
	64	+	}
	65	+
	66	+	// ClearDeployEventSuppression removes event suppression for an image.
	67	+	func (s *Service) ClearDeployEventSuppression(imageName string) {
	68	+	if v, loaded := s.suppressedImages.LoadAndDelete(imageName); loaded {
	69	+	v.(*time.Timer).Stop()
	70	+	}
	71	+	}
	72	+
	73	+	// IsDeployEventSuppressed checks if deploy events are suppressed for an image.
	74	+	func (s *Service) IsDeployEventSuppressed(imageName string) bool {
	75	+	_, exists := s.suppressedImages.Load(imageName)
	76	+	return exists
	77	+	}
	78	+
51	79		// GetManifest retrieves a manifest by name and reference.
52	80		func (s Service) GetManifest(ctx context.Context, name, reference string) (domain.Manifest, error) {
53	81		ctx = zerowrap.CtxWithFields(ctx, map[string]any{

@@ -110,13 +138,17 @@ func (s Service) PutManifest(ctx context.Context, manifest domain.Manifest) (s
110	138		// A docker push sends manifests by both digest and tag; firing only on
111	139		// tag prevents duplicate deploy triggers for the same push.
112	140		if s.eventBus != nil && !strings.HasPrefix(manifest.Reference, "sha256:") {
113		-	if err := s.eventBus.Publish(domain.EventImagePushed, domain.ImagePushedPayload{
114		-	Name: manifest.Name,
115		-	Reference: manifest.Reference,
116		-	Manifest: manifest.Data,
117		-	Annotations: manifest.Annotations,
118		-	}); err != nil {
119		-	log.Warn().Err(err).Msg("failed to publish image pushed event")
	141	+	if s.IsDeployEventSuppressed(manifest.Name) {
	142	+	log.Info().Str("image", manifest.Name).Msg("skipping image.pushed event: CLI deploy intent active")
	143	+	} else {
	144	+	if err := s.eventBus.Publish(domain.EventImagePushed, domain.ImagePushedPayload{
	145	+	Name: manifest.Name,
	146	+	Reference: manifest.Reference,
	147	+	Manifest: manifest.Data,
	148	+	Annotations: manifest.Annotations,
	149	+	}); err != nil {
	150	+	log.Warn().Err(err).Msg("failed to publish image pushed event")
	151	+	}
120	152		}
121	153		}
122	154

 	"fmt"
 	"io"
 	"strings"
+	"sync"
+	"time"
 	"github.com/bnema/zerowrap"
 	"go.opentelemetry.io/otel"
 // Service implements the RegistryService interface.
 type Service struct {
-	blobStorage     out.BlobStorage
+	blobStorage      out.BlobStorage
-	manifestStorage out.ManifestStorage
+	manifestStorage  out.ManifestStorage
-	eventBus        out.EventPublisher
+	eventBus         out.EventPublisher
-	metrics         *telemetry.Metrics
+	metrics          *telemetry.Metrics
+	suppressedImages sync.Map // imageName -> *time.Timer
 }
 // SetMetrics sets the telemetry metrics for the registry service.
 	}
 }
+// SuppressDeployEvent marks an image name to skip image.pushed events.
+// The suppression auto-expires after 2 minutes to prevent leaks.
+func (s *Service) SuppressDeployEvent(imageName string) {
+	timer := time.AfterFunc(2*time.Minute, func() {
+		s.suppressedImages.Delete(imageName)
+	})
+	if existing, loaded := s.suppressedImages.LoadOrStore(imageName, timer); loaded {
+		existing.(*time.Timer).Stop()
+		s.suppressedImages.Store(imageName, timer)
+	}
+}
+// ClearDeployEventSuppression removes event suppression for an image.
+func (s *Service) ClearDeployEventSuppression(imageName string) {
+	if v, loaded := s.suppressedImages.LoadAndDelete(imageName); loaded {
+		v.(*time.Timer).Stop()
+	}
+}
+// IsDeployEventSuppressed checks if deploy events are suppressed for an image.
+func (s *Service) IsDeployEventSuppressed(imageName string) bool {
+	_, exists := s.suppressedImages.Load(imageName)
+	return exists
+}
 // GetManifest retrieves a manifest by name and reference.
 func (s *Service) GetManifest(ctx context.Context, name, reference string) (*domain.Manifest, error) {
 	ctx = zerowrap.CtxWithFields(ctx, map[string]any{
 	// A docker push sends manifests by both digest and tag; firing only on
 	// tag prevents duplicate deploy triggers for the same push.
 	if s.eventBus != nil && !strings.HasPrefix(manifest.Reference, "sha256:") {
-		if err := s.eventBus.Publish(domain.EventImagePushed, domain.ImagePushedPayload{
+		if s.IsDeployEventSuppressed(manifest.Name) {
-			Name:        manifest.Name,
+			log.Info().Str("image", manifest.Name).Msg("skipping image.pushed event: CLI deploy intent active")
-			Reference:   manifest.Reference,
+		} else {
-			Manifest:    manifest.Data,
+			if err := s.eventBus.Publish(domain.EventImagePushed, domain.ImagePushedPayload{
-			Annotations: manifest.Annotations,
+				Name:        manifest.Name,
-		}); err != nil {
+				Reference:   manifest.Reference,
-			log.Warn().Err(err).Msg("failed to publish image pushed event")
+				Manifest:    manifest.Data,
+				Annotations: manifest.Annotations,
+			}); err != nil {
+				log.Warn().Err(err).Msg("failed to publish image pushed event")
+			}
 		}
 	}

M internal/usecase/registry/service_test.go +39 -0


@@ -11,6 +11,7 @@ import (
11	11		"github.com/bnema/zerowrap"
12	12		"github.com/stretchr/testify/assert"
13	13		"github.com/stretchr/testify/mock"
	14	+	"github.com/stretchr/testify/require"
14	15
15	16		"github.com/bnema/gordon/internal/boundaries/out/mocks"
16	17		"github.com/bnema/gordon/internal/domain"

@@ -437,3 +438,41 @@ func TestService_ListRepositories_Error(t *testing.T) {
437	438		assert.Nil(t, repos)
438	439		assert.Contains(t, err.Error(), "failed to list repositories")
439	440		}
	441	+
	442	+	func TestPutManifest_SkipsEventWhenDeployIntentSuppressed(t *testing.T) {
	443	+	blobStorage := mocks.NewMockBlobStorage(t)
	444	+	manifestStorage := mocks.NewMockManifestStorage(t)
	445	+	eventBus := mocks.NewMockEventPublisher(t)
	446	+
	447	+	svc := NewService(blobStorage, manifestStorage, eventBus)
	448	+
	449	+	svc.SuppressDeployEvent("my-app")
	450	+
	451	+	manifestStorage.EXPECT().PutManifest("my-app", "latest", "application/vnd.oci.image.manifest.v1+json", mock.Anything).Return(nil)
	452	+
	453	+	manifest := &domain.Manifest{
	454	+	Name: "my-app",
	455	+	Reference: "latest",
	456	+	ContentType: "application/vnd.oci.image.manifest.v1+json",
	457	+	Data: []byte(`{"schemaVersion":2}`),
	458	+	}
	459	+
	460	+	_, err := svc.PutManifest(context.Background(), manifest)
	461	+	require.NoError(t, err)
	462	+
	463	+	eventBus.AssertNotCalled(t, "Publish", mock.Anything, mock.Anything)
	464	+	}
	465	+
	466	+	func TestSuppressDeployEvent_ClearsCorrectly(t *testing.T) {
	467	+	blobStorage := mocks.NewMockBlobStorage(t)
	468	+	manifestStorage := mocks.NewMockManifestStorage(t)
	469	+	eventBus := mocks.NewMockEventPublisher(t)
	470	+
	471	+	svc := NewService(blobStorage, manifestStorage, eventBus)
	472	+
	473	+	svc.SuppressDeployEvent("my-app")
	474	+	assert.True(t, svc.IsDeployEventSuppressed("my-app"))
	475	+
	476	+	svc.ClearDeployEventSuppression("my-app")
	477	+	assert.False(t, svc.IsDeployEventSuppressed("my-app"))
	478	+	}


@@ -11,6 +11,7 @@ import (
11	"github.com/bnema/zerowrap"	11		"github.com/bnema/zerowrap"
12	"github.com/stretchr/testify/assert"	12		"github.com/stretchr/testify/assert"
13	"github.com/stretchr/testify/mock"	13		"github.com/stretchr/testify/mock"
		14	+	"github.com/stretchr/testify/require"
14		15
15	"github.com/bnema/gordon/internal/boundaries/out/mocks"	16		"github.com/bnema/gordon/internal/boundaries/out/mocks"
16	"github.com/bnema/gordon/internal/domain"	17		"github.com/bnema/gordon/internal/domain"

@@ -437,3 +438,41 @@ func TestService_ListRepositories_Error(t *testing.T) {
437	assert.Nil(t, repos)	438		assert.Nil(t, repos)
438	assert.Contains(t, err.Error(), "failed to list repositories")	439		assert.Contains(t, err.Error(), "failed to list repositories")
439	}	440		}
		441	+
		442	+	func TestPutManifest_SkipsEventWhenDeployIntentSuppressed(t *testing.T) {
		443	+	blobStorage := mocks.NewMockBlobStorage(t)
		444	+	manifestStorage := mocks.NewMockManifestStorage(t)
		445	+	eventBus := mocks.NewMockEventPublisher(t)
		446	+
		447	+	svc := NewService(blobStorage, manifestStorage, eventBus)
		448	+
		449	+	svc.SuppressDeployEvent("my-app")
		450	+
		451	+	manifestStorage.EXPECT().PutManifest("my-app", "latest", "application/vnd.oci.image.manifest.v1+json", mock.Anything).Return(nil)
		452	+
		453	+	manifest := &domain.Manifest{
		454	+	Name: "my-app",
		455	+	Reference: "latest",
		456	+	ContentType: "application/vnd.oci.image.manifest.v1+json",
		457	+	Data: []byte(`{"schemaVersion":2}`),
		458	+	}
		459	+
		460	+	_, err := svc.PutManifest(context.Background(), manifest)
		461	+	require.NoError(t, err)
		462	+
		463	+	eventBus.AssertNotCalled(t, "Publish", mock.Anything, mock.Anything)
		464	+	}
		465	+
		466	+	func TestSuppressDeployEvent_ClearsCorrectly(t *testing.T) {
		467	+	blobStorage := mocks.NewMockBlobStorage(t)
		468	+	manifestStorage := mocks.NewMockManifestStorage(t)
		469	+	eventBus := mocks.NewMockEventPublisher(t)
		470	+
		471	+	svc := NewService(blobStorage, manifestStorage, eventBus)
		472	+
		473	+	svc.SuppressDeployEvent("my-app")
		474	+	assert.True(t, svc.IsDeployEventSuppressed("my-app"))
		475	+
		476	+	svc.ClearDeployEventSuppression("my-app")
		477	+	assert.False(t, svc.IsDeployEventSuppressed("my-app"))
		478	+	}