go/runtime/registry: Cleanup bundles for versions lower then active

.changelog/5737.feature.md

@@ -0,0 +1 @@
+go/runtime/bundle: Cleanup bundles on runtime upgrade

go/runtime/bundle/manager.go

@@ -19,6 +19,7 @@ import (
 	"github.com/oasisprotocol/oasis-core/go/common/crypto/hash"
 	"github.com/oasisprotocol/oasis-core/go/common/logging"
 	cmSync "github.com/oasisprotocol/oasis-core/go/common/sync"
+	"github.com/oasisprotocol/oasis-core/go/common/version"
 	"github.com/oasisprotocol/oasis-core/go/config"
 )
 
@@ -46,6 +47,12 @@ type ManifestStore interface {
 	// AddManifest adds the provided manifest, whose components were extracted
 	// to the specified directory, to the store.
 	AddManifest(manifest *Manifest, dir string) error
+
+	// RemoveManifests removes all regular manifests for the specified runtimeID
+	// whose versions are lower than the provided version.
+	//
+	// Returns a list of directories for the removed exploded bundles.
+	RemoveManifests(runtimeID common.Namespace, version version.Version) ([]string, error)
 }
 
 // Manager is responsible for managing bundles.
@@ -62,8 +69,9 @@ type Manager struct {
 	runtimeBaseURLs map[common.Namespace][]string
 	globalBaseURLs  []string
 
-	downloadCh    chan struct{}
-	downloadQueue map[common.Namespace][]hash.Hash
+	downloadAndCleanCh chan struct{}
+	downloadQueue      map[common.Namespace][]hash.Hash
+	cleanupQueue       map[common.Namespace]version.Version
 
 	client *http.Client
 	store  ManifestStore
@@ -119,8 +127,9 @@ func NewManager(dataDir string, runtimeIDs []common.Namespace, store ManifestSto
 		runtimeIDs:         runtimes,
 		globalBaseURLs:     globalBaseURLs,
 		runtimeBaseURLs:    runtimeBaseURLs,
-		downloadCh:         make(chan struct{}, 1),
+		downloadAndCleanCh: make(chan struct{}, 1),
 		downloadQueue:      make(map[common.Namespace][]hash.Hash),
+		cleanupQueue:       make(map[common.Namespace]version.Version),
 		client:             &client,
 		store:              store,
 		logger:             *logger,
@@ -192,12 +201,13 @@ func (m *Manager) run(ctx context.Context) {
 	for {
 		select {
 		case <-ticker.C:
-		case <-m.downloadCh:
+		case <-m.downloadAndCleanCh:
 		case <-ctx.Done():
 			m.logger.Info("stopping")
 			return
 		}
 
+		m.Clean()
 		m.download()
 	}
 }
@@ -236,9 +246,25 @@ func (m *Manager) Download(runtimeID common.Namespace, manifestHashes []hash.Has
 	}
 	m.downloadQueue[runtimeID] = hashes
 
-	// Trigger immediate download of new bundles.
+	// Trigger immediate download and clean-up of stale bundles.
+	select {
+	case m.downloadAndCleanCh <- struct{}{}:
+	default:
+	}
+}
+
+// Cleanup initiates the clean-up of regular bundles for the specified runtime,
+// removing versions lower than the specified one.
+func (m *Manager) Cleanup(runtimeID common.Namespace, active version.Version) {
+	// Update the active versions.
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	m.cleanupQueue[runtimeID] = active
+
+	// Trigger immediate download and clean-up of stale bundles.
 	select {
-	case m.downloadCh <- struct{}{}:
+	case m.downloadAndCleanCh <- struct{}{}:
 	default:
 	}
 }
@@ -250,6 +276,15 @@ func (m *Manager) download() {
 	}
 }
 
+// Clean removes outdated manifest hashes and deletes corresponding
+// exploded bundles for runtimes in the clean-up queue.
+func (m *Manager) Clean() {
+	m.logger.Info("removing regular bundles")
+	for runtimeID := range m.runtimeIDs {
+		m.cleanStaleBundles(runtimeID)
+	}
+}
+
 func (m *Manager) downloadBundles(runtimeID common.Namespace) {
 	// Try to download queued bundles.
 	m.mu.RLock()
@@ -437,6 +472,32 @@ func (m *Manager) fetchBundle(url string) (string, error) {
 	return file.Name(), nil
 }
 
+func (m *Manager) cleanStaleBundles(runtimeID common.Namespace) {
+	m.mu.Lock()
+	active, ok := m.cleanupQueue[runtimeID]
+	m.mu.Unlock()
+
+	if !ok {
+		return
+	}
+
+	// TODO should you also remove dirs for manifests successfully removed,
+	// even if error?
+	dirs, err := m.store.RemoveManifests(runtimeID, active)
+	if err != nil {
+		m.logger.Error("failed to remove regular manifests from the registry",
+			"runtimeID", runtimeID,
+			"version", active,
+			"err", err,
+		)
+		return
+	}
+
+	for _, dir := range dirs {
+		_ = m.removeBundle(dir)
+	}
+}
+
 func (m *Manager) loadManifests() (map[string]*Manifest, error) {
 	m.logger.Info("loading manifests")
 

go/runtime/bundle/manager_test.go

@@ -5,7 +5,9 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	"github.com/oasisprotocol/oasis-core/go/common"
 	"github.com/oasisprotocol/oasis-core/go/common/crypto/hash"
+	"github.com/oasisprotocol/oasis-core/go/common/version"
 )
 
 type mockStore struct {
@@ -28,6 +30,10 @@ func (r *mockStore) AddManifest(manifest *Manifest, _ string) error {
 	return nil
 }
 
+func (r *mockStore) RemoveManifests(_ common.Namespace, _ version.Version) ([]string, error) {
+	panic("RemoveManifests not implemented")
+}
+
 func TestRegisterManifest(t *testing.T) {
 	store := newMockStore()
 	manager, err := NewManager("", nil, store)

go/runtime/bundle/registry.go

@@ -1,6 +1,7 @@
 package bundle
 
 import (
+	"errors"
 	"fmt"
 	"maps"
 	"slices"
@@ -152,6 +153,60 @@ func (r *Registry) AddManifest(manifest *Manifest, dir string) error {
 	return nil
 }
 
+// RemoveManifests removes all regular manifests for the specified runtime ID
+// whose versions are lower than the provided version.
+//
+// Returns a list of directories for the removed exploded bundles.
+func (r *Registry) RemoveManifests(runtimeID common.Namespace, version version.Version) ([]string, error) {
+	r.logger.Info("removing manifests with versions lower then provided",
+		"id", runtimeID,
+		"version", version,
+	)
+
+	var stale []string
+	for _, v := range r.GetVersions(runtimeID) {
+		if !v.Less(version) {
+			continue
+		}
+
+		dir, err := r.removeManifest(runtimeID, v)
+		if err != nil {
+			return nil, err
+		}
+		stale = append(stale, dir)
+	}
+	return stale, nil
+}
+
+func (r *Registry) removeManifest(runtimeID common.Namespace, version version.Version) (string, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	var explDir string
+	manifest, ok := r.regularManifests[runtimeID][version]
+	if !ok {
+		return explDir, fmt.Errorf("missing regular manifest for runtime ID %v with %s", runtimeID, version.String())
+	}
+	explComp := r.components[runtimeID][component.ID_RONL][version]
+	if explComp == nil {
+		return explDir, errors.New("components missing RONL component")
+	}
+	explDir = explComp.ExplodedDataDir
+	manifestHash := manifest.Hash()
+
+	r.logger.Info("Removing (regular) manifest from registry",
+		"id", runtimeID,
+		"version", version,
+		"manifest_hash", manifestHash,
+	)
+
+	delete(r.regularManifests[runtimeID], version)
+	delete(r.manifestHashes, manifestHash)
+	for _, c := range manifest.Components {
+		delete(r.components[runtimeID][c.ID()], c.Version)
+	}
+	return explDir, nil
+}
+
 // GetVersions returns versions for the given runtime, sorted in ascending
 // order.
 func (r *Registry) GetVersions(runtimeID common.Namespace) []version.Version {

go/runtime/registry/registry.go

@@ -388,11 +388,26 @@ func (r *runtime) run(ctx context.Context) {
 		select {
 		case <-ctx.Done():
 			return
-		case <-epoCh:
+		case epoch := <-epoCh:
 			if up := r.updateActiveDescriptor(ctx); up && !activeInitialized {
 				close(r.activeDescriptorCh)
 				activeInitialized = true
 			}
+
+			// Trigger clean-up for bundles less than active version.
+			r.RLock()
+			rt := r.activeDescriptor
+			r.RUnlock()
+			if rt == nil {
+				continue
+			}
+
+			active := rt.ActiveDeployment(epoch)
+			if active == nil {
+				continue
+			}
+
+			r.bundleManager.Cleanup(rt.ID, active.Version)
 		case rt := <-regCh:
 			if !rt.ID.Equal(&r.id) {
 				continue