Send an async update notification to the next latest revision if the existing latest package revision is deleted (#380)

* Send an update notification after the latest package revision is deleted

* Send async notification in crcache when latest pkg revision is deleted

* Add tracing to crcache sendLatestPkgUpdateNotification

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,31 @@ 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) {
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
+	defer cancel()
+	_, span := tracer.Start(ctx, "cachedRepository::sendLatestPkgUpdateNotification", trace.WithAttributes())
+	defer span.End()
+	// 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

@@ -25,6 +25,7 @@ import (
 	"go.opentelemetry.io/otel/trace"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/klog/v2"
 )
 
@@ -134,9 +135,35 @@ func (p *dbPackage) DeletePackageRevision(ctx context.Context, old repository.Pa
 		return pkgDeleteFromDB(ctx, p.Key())
 	}
 
+	if dbPR.IsLatestRevision() {
+		klog.Infof("dbPackage %+v: latest PackageRevision deleted. Sending notification.", p.Key())
+		go p.sendLatestPkgUpdateNotification()
+	}
+
 	return nil
 }
 
+// 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())
+	if err != nil {
+		klog.Error(err)
+		return
+	} else if latestRevision == nil {
+		klog.Infof("dbPackage %+v: no new latest PackageRevision found. Notification not sent.", p.Key())
+		return
+	}
+
+	sent := p.repo.repoPRChangeNotifier.NotifyPackageRevisionChange(watch.Modified, latestRevision)
+	klog.Infof("dbcache: sent %d for latest PackageRevision %s/%s", sent, latestRevision.KubeObjectNamespace(), latestRevision.KubeObjectName())
+
+}
+
 func (p *dbPackage) GetLatestRevision(ctx context.Context) int {
 	_, span := tracer.Start(ctx, "dbPackage:GetLatestRevision", trace.WithAttributes())
 	defer span.End()

pkg/cache/dbcache/dbpackagerevision_test.go

@@ -17,6 +17,7 @@ package dbcache
 import (
 	"context"
 	"errors"
+	"time"
 
 	porchapi "github.com/nephio-project/porch/api/porch/v1alpha1"
 	configapi "github.com/nephio-project/porch/api/porchconfig/v1alpha1"
@@ -310,3 +311,97 @@ type fakeRepoWithDeleteError struct {
 func (r *fakeRepoWithDeleteError) DeletePackageRevision(context.Context, repository.PackageRevision) error {
 	return errors.New("package not found")
 }
+
+func (t *DbTestSuite) TestDBDeleteLatestRevision() {
+	// Test that the async notification logic is triggered
+	ctx := t.Context()
+
+	mockCache := mockcachetypes.NewMockCache(t.T())
+	cachetypes.CacheInstance = mockCache
+
+	dbRepo := t.createTestRepo("test-ns", "test-repo")
+	mockCache.EXPECT().GetRepository(mock.Anything).Return(dbRepo)
+	dbPkg := t.createTestPkg(dbRepo.Key(), "test-package")
+	dbPkg.repo = dbRepo
+
+	// Create and write first package revision to database
+	dbPR1 := dbPackageRevision{
+		repo: dbRepo,
+		pkgRevKey: repository.PackageRevisionKey{
+			PkgKey:        dbPkg.Key(),
+			WorkspaceName: "workspace-1",
+			Revision:      1,
+		},
+		meta:      metav1.ObjectMeta{},
+		spec:      &porchapi.PackageRevisionSpec{},
+		updated:   time.Now().UTC(),
+		updatedBy: "testuser",
+		lifecycle: porchapi.PackageRevisionLifecyclePublished,
+		latest:    false,
+		resources: map[string]string{},
+	}
+
+	// Create and write main package revision to make sure len(prSlice) > 0
+	dbPRMain := dbPackageRevision{
+		repo: dbRepo,
+		pkgRevKey: repository.PackageRevisionKey{
+			PkgKey:        dbPkg.Key(),
+			WorkspaceName: "main",
+			Revision:      -1,
+		},
+		meta:      metav1.ObjectMeta{},
+		spec:      &porchapi.PackageRevisionSpec{},
+		updated:   time.Now().UTC(),
+		updatedBy: "testuser",
+		lifecycle: porchapi.PackageRevisionLifecyclePublished,
+		resources: map[string]string{},
+	}
+
+	err := pkgRevWriteToDB(ctx, &dbPR1)
+	t.Require().NoError(err)
+
+	err = pkgRevWriteToDB(ctx, &dbPRMain)
+	t.Require().NoError(err)
+
+	// Create and write second package revision to database (this will be latest)
+	dbPR2 := dbPackageRevision{
+		repo: dbRepo,
+		pkgRevKey: repository.PackageRevisionKey{
+			PkgKey:        dbPkg.Key(),
+			WorkspaceName: "workspace-2",
+			Revision:      2,
+		},
+		meta:      metav1.ObjectMeta{},
+		spec:      &porchapi.PackageRevisionSpec{},
+		updated:   time.Now().UTC(),
+		updatedBy: "testuser",
+		lifecycle: porchapi.PackageRevisionLifecyclePublished,
+		latest:    true,
+		resources: map[string]string{},
+	}
+
+	err = pkgRevWriteToDB(ctx, &dbPR2)
+	t.Require().NoError(err)
+
+	// Delete the latest revision - should trigger async notification
+	err = dbPkg.DeletePackageRevision(ctx, &dbPR2, false)
+	t.Require().NoError(err)
+
+	// wait for async call to finish
+	time.Sleep(100 * time.Millisecond)
+
+	dbPR1.latest = true
+	err = dbPkg.DeletePackageRevision(ctx, &dbPR1, false)
+	t.Require().NoError(err)
+
+	// wait for async call to finish
+	time.Sleep(100 * time.Millisecond)
+
+	// After deletion, dbPRMain should remain in database so len(prSlice) > 0
+	err = dbPkg.DeletePackageRevision(ctx, &dbPRMain, false)
+	t.Require().NoError(err)
+
+	// Clean up
+	err = repoDeleteFromDB(ctx, dbRepo.Key())
+	t.Require().NoError(err)
+}