Send async notification in crcache when latest pkg revision is deleted

Author: kushnaidu

pkg/cache/crcache/repository.go

@@ -405,6 +405,9 @@ func (r *cachedRepository) DeletePackageRevision(ctx context.Context, prToDelete
 	}
 	klog.Infof("PackageRevision %s deleted for real since no finalizers", prToDelete.KubeObjectName())
 
+	// Check if this is the latest revision before deletion
+	isLatest := prToDelete.(*cachedPackageRevision).IsLatestRevision()
+
 	// Unwrap
 	unwrapped := prToDelete.(*cachedPackageRevision).PackageRevision
 	if err := r.repo.DeletePackageRevision(ctx, unwrapped); err != nil {
@@ -421,6 +424,12 @@ func (r *cachedRepository) DeletePackageRevision(ctx context.Context, prToDelete
 		identifyLatestRevisions(ctx, r.cachedPackageRevisions)
 	}
 
+	// Check if we need to send async notification for new latest revision
+	if isLatest {
+		klog.Infof("crcache: %+v: latest PackageRevision deleted. Sending notification.", prToDelete.Key().PkgKey)
+		go r.sendLatestPkgUpdateNotification(prToDelete.Key().PkgKey)
+	}
+
 	r.mutex.Unlock()
 
 	if _, err := r.metadataStore.Delete(ctx, namespacedName, true); err != nil {
@@ -436,6 +445,27 @@ func (r *cachedRepository) DeletePackageRevision(ctx context.Context, prToDelete
 	return nil
 }
 
+// sendLatestPkgUpdateNotification sends async notification when a new latest package revision is identified
+func (r *cachedRepository) sendLatestPkgUpdateNotification(pkgKey repository.PackageKey) {
+	// Find the new latest revision for this package
+	r.mutex.RLock()
+	var newLatest repository.PackageRevision
+	for _, pr := range r.cachedPackageRevisions {
+		if pr.Key().PkgKey == pkgKey && pr.IsLatestRevision() {
+			newLatest = pr
+			break
+		}
+	}
+	r.mutex.RUnlock()
+
+	if newLatest != nil {
+		sent := r.repoPRChangeNotifier.NotifyPackageRevisionChange(watch.Modified, newLatest)
+		klog.Infof("crcache: async notification sent %d for new latest PackageRevision %s/%s", sent, newLatest.KubeObjectNamespace(), newLatest.KubeObjectName())
+	} else {
+		klog.Infof("crcache: no new latest revision found for package %s after deletion. Notification not sent.", pkgKey.Package)
+	}
+}
+
 func (r *cachedRepository) ListPackages(ctx context.Context, filter repository.ListPackageFilter) ([]repository.Package, error) {
 	packages, err := r.getPackages(ctx, filter, false)
 	if err != nil {

pkg/cache/crcache/repository_test.go

@@ -34,6 +34,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/watch"
 )
 
 func TestCachedRepoRefresh(t *testing.T) {
@@ -250,3 +251,111 @@ func TestHandleRunOnceAt(t *testing.T) {
 	assert.NotNil(t, status, "Expected repository status to be updated")
 	assert.Contains(t, []string{"Ready", "Error", "Reconciling"}, status.Conditions[0].Reason)
 }
+
+func TestCRDeleteLatestRevision(t *testing.T) {
+	mockRepo := mockrepo.NewMockRepository(t)
+	mockMeta := mockmeta.NewMockMetadataStore(t)
+	mockNotifier := mockcachetypes.NewMockRepoPRChangeNotifier(t)
+
+	repoSpec := configapi.Repository{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      repoName,
+			Namespace: namespace,
+		},
+	}
+
+	repoKey := repository.RepositoryKey{
+		Namespace: namespace,
+		Name:      repoName,
+	}
+
+	// Create two package revisions for the same package
+	prKey1 := repository.PackageRevisionKey{
+		PkgKey: repository.PackageKey{
+			RepoKey: repoKey,
+			Path:    "",
+			Package: "test-package",
+		},
+		WorkspaceName: "v1",
+		Revision:      1,
+	}
+
+	prKey2 := repository.PackageRevisionKey{
+		PkgKey: repository.PackageKey{
+			RepoKey: repoKey,
+			Path:    "",
+			Package: "test-package",
+		},
+		WorkspaceName: "v2",
+		Revision:      2,
+	}
+
+	fpr1 := &fake.FakePackageRevision{
+		PrKey:            prKey1,
+		PackageLifecycle: porchapi.PackageRevisionLifecyclePublished,
+	}
+
+	fpr2 := &fake.FakePackageRevision{
+		PrKey:            prKey2,
+		PackageLifecycle: porchapi.PackageRevisionLifecyclePublished,
+	}
+
+	// Setup mocks
+	mockRepo.On("Key").Return(repoKey).Maybe()
+	mockRepo.EXPECT().DeletePackageRevision(mock.Anything, mock.Anything).Return(nil).Maybe()
+	mockMeta.EXPECT().Delete(mock.Anything, mock.Anything, mock.Anything).Return(metav1.ObjectMeta{}, nil).Maybe()
+
+	// Expect exactly 2 notifications: one for deletion, one for async notification
+	mockNotifier.EXPECT().NotifyPackageRevisionChange(watch.Deleted, mock.Anything).Return(1).Once()
+	mockNotifier.EXPECT().NotifyPackageRevisionChange(watch.Modified, mock.Anything).Return(1).Once()
+
+	// Create repository manually to avoid sync manager
+	cr := &cachedRepository{
+		key:                  repoKey,
+		repoSpec:             &repoSpec,
+		repo:                 mockRepo,
+		metadataStore:        mockMeta,
+		repoPRChangeNotifier: mockNotifier,
+		cachedPackages:       make(map[repository.PackageKey]*cachedPackage), // Initialize to enable deletion
+	}
+
+	// Initialize cache with two revisions
+	cr.cachedPackageRevisions = make(map[repository.PackageRevisionKey]*cachedPackageRevision)
+	cr.cachedPackageRevisions[prKey1] = &cachedPackageRevision{
+		PackageRevision:  fpr1,
+		metadataStore:    mockMeta,
+		isLatestRevision: false,
+	}
+	cr.cachedPackageRevisions[prKey2] = &cachedPackageRevision{
+		PackageRevision:  fpr2,
+		metadataStore:    mockMeta,
+		isLatestRevision: false,
+	}
+
+	// Identify latest revisions (revision 2 should be latest)
+	identifyLatestRevisions(context.TODO(), cr.cachedPackageRevisions)
+
+	// Verify revision 2 is marked as latest before deletion
+	assert.True(t, cr.cachedPackageRevisions[prKey2].IsLatestRevision())
+	assert.False(t, cr.cachedPackageRevisions[prKey1].IsLatestRevision())
+
+	// Delete the latest revision (revision 2)
+	err := cr.DeletePackageRevision(context.TODO(), cr.cachedPackageRevisions[prKey2])
+	assert.NoError(t, err)
+
+	// Give time for async notification to complete
+	time.Sleep(100 * time.Millisecond)
+
+	// Now delete pkg1 - should only send 1 notification (deletion) since no more revisions remain
+	mockNotifier.EXPECT().NotifyPackageRevisionChange(watch.Deleted, mock.Anything).Return(1).Once()
+	// No async notification expected since no revisions remain
+
+	err = cr.DeletePackageRevision(context.TODO(), cr.cachedPackageRevisions[prKey1])
+	assert.NoError(t, err)
+
+	// Give time for any potential async notification
+	time.Sleep(100 * time.Millisecond)
+
+	// Verify all expected mock calls were made
+	mockNotifier.AssertExpectations(t)
+}

pkg/cache/dbcache/dbpackage.go

@@ -137,16 +137,17 @@ func (p *dbPackage) DeletePackageRevision(ctx context.Context, old repository.Pa
 
 	if dbPR.IsLatestRevision() {
 		klog.Infof("dbPackage %+v: latest PackageRevision deleted. Sending notification.", p.Key())
-		updateCtx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
-		defer cancel()
-		go p.sendLatestPkgUpdateNotification(updateCtx)
+		go p.sendLatestPkgUpdateNotification()
 	}
 
 	return nil
 }
 
-func (p *dbPackage) sendLatestPkgUpdateNotification(ctx context.Context) {
-	_, span := tracer.Start(ctx, "dbPackage:sendLatestPkgUpdateNotification", trace.WithAttributes())
+// sendLatestPkgUpdateNotification sends async notification when a new latest package revision is identified
+func (p *dbPackage) sendLatestPkgUpdateNotification() {
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
+	defer cancel()
+	_, span := tracer.Start(ctx, "dbPackage::sendLatestPkgUpdateNotification", trace.WithAttributes())
 	defer span.End()
 
 	latestRevision, err := pkgRevReadLatestPRFromDB(ctx, p.Key())

pkg/cache/dbcache/dbpackagerevision_test.go

@@ -312,7 +312,7 @@ func (r *fakeRepoWithDeleteError) DeletePackageRevision(context.Context, reposit
 	return errors.New("package not found")
 }
 
-func (t *DbTestSuite) TestDBPackageRevisionDeleteLatest() {
+func (t *DbTestSuite) TestDBDeleteLatestRevision() {
 	// Test that the async notification logic is triggered
 	ctx := t.Context()