[evict] Optimize performance by deleting expired entities in single statements and remove the listing limits

cmds/db-manager/cleanup/evict.go

@@ -8,14 +8,8 @@ import (
 
 	"github.com/interuss/dss/pkg/logging"
 	dssmodels "github.com/interuss/dss/pkg/models"
-	ridmodels "github.com/interuss/dss/pkg/rid/models"
-	ridrepos "github.com/interuss/dss/pkg/rid/repos"
 	rids "github.com/interuss/dss/pkg/rid/store"
-	scdmodels "github.com/interuss/dss/pkg/scd/models"
-	scdrepos "github.com/interuss/dss/pkg/scd/repos"
 	scds "github.com/interuss/dss/pkg/scd/store"
-	dssstore "github.com/interuss/dss/pkg/store"
-	"github.com/interuss/stacktrace"
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
 )
@@ -66,101 +60,92 @@ func evict(cmd *cobra.Command, _ []string) error {
 	}
 
 	var (
-		expiredOpIntents []*scdmodels.OperationalIntent
-		scdExpiredSub    []*scdmodels.Subscription
-		expiredISAs      []*ridmodels.IdentificationServiceArea
-		ridExpiredSub    []*ridmodels.Subscription
+		expiredOpIntents []dssmodels.ID
+		scdExpiredSub    []dssmodels.ID
+		expiredISAs      []dssmodels.ID
+		ridExpiredSub    []dssmodels.ID
 	)
-	scdAction := func(ctx context.Context, r scdrepos.Repository) (err error) {
-		if *checkScdOirs {
-			expiredOpIntents, err = r.ListExpiredOperationalIntents(ctx, scdThreshold)
+	scdRepo, err := scdStore.Interact(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to interact with SCD store: %w", err)
+	}
+
+	if *checkScdOirs {
+		if *deleteExpired {
+			expiredOpIntents, err = scdRepo.DeleteExpiredOperationalIntents(ctx, scdThreshold)
 			if err != nil {
-				return fmt.Errorf("listing expired operational intents: %w", err)
+				return fmt.Errorf("failed to delete expired operational intents: %w", err)
 			}
-			if *deleteExpired {
-				for _, opIntent := range expiredOpIntents {
-					if err = r.DeleteOperationalIntent(ctx, opIntent.ID); err != nil {
-						return fmt.Errorf("deleting expired operational intents: %w", err)
-					}
-				}
+		} else {
+			expiredOpIntents, err = scdRepo.ListExpiredOperationalIntents(ctx, scdThreshold)
+			if err != nil {
+				return fmt.Errorf("failed to list expired operational intents: %w", err)
 			}
 		}
+	}
 
-		if *checkScdSubs {
-			scdExpiredSub, err = r.ListExpiredSubscriptions(ctx, scdThreshold)
+	if *checkScdSubs {
+		if *deleteExpired {
+			scdExpiredSub, err = scdRepo.DeleteExpiredSubscriptions(ctx, scdThreshold)
 			if err != nil {
-				return fmt.Errorf("SCD listing expired subscriptions: %w", err)
+				return fmt.Errorf("failed to delete expired SCD subscriptions: %w", err)
 			}
-			if *deleteExpired {
-				for _, sub := range scdExpiredSub {
-					if err = r.DeleteSubscription(ctx, sub.ID); err != nil {
-						return fmt.Errorf("SCD deleting expired subscriptions: %w", err)
-					}
-				}
+		} else {
+			scdExpiredSub, err = scdRepo.ListExpiredSubscriptions(ctx, scdThreshold)
+			if err != nil {
+				return fmt.Errorf("failed to list expired SCD subscriptions: %w", err)
 			}
 		}
-		return nil
-	}
-	if _, err = scdStore.Transact(ctx, dssstore.NewFuncOperation(scdAction)); err != nil {
-		return fmt.Errorf("failed to execute SCD transaction: %w", err)
 	}
 
-	ridAction := func(ctx context.Context, r ridrepos.Repository) (err error) {
-		if *checkRidISAs {
+	ridRepo, err := ridStore.Interact(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to interact with RID store: %w", err)
+	}
 
-			expiredISAs, err = r.ListExpiredISAs(ctx, *locality, ridThreshold)
+	if *checkRidISAs {
+		if *deleteExpired {
+			expiredISAs, err = ridRepo.DeleteExpiredISAs(ctx, *locality, ridThreshold)
 			if err != nil {
-				return stacktrace.Propagate(err, "Failed to list expired ISAs")
+				return fmt.Errorf("failed to delete expired ISAs: %w", err)
 			}
-
-			if *deleteExpired {
-				for _, isa := range expiredISAs {
-					_, err := r.DeleteISA(ctx, isa)
-					if err != nil {
-						return stacktrace.Propagate(err, "Failed to delete ISAs")
-					}
-				}
+		} else {
+			expiredISAs, err = ridRepo.ListExpiredISAs(ctx, *locality, ridThreshold)
+			if err != nil {
+				return fmt.Errorf("failed to list expired ISAs: %w", err)
 			}
-
 		}
+	}
 
-		if *checkRidSubs {
-
-			ridExpiredSub, err = r.ListExpiredSubscriptions(ctx, *locality, ridThreshold)
+	if *checkRidSubs {
+		if *deleteExpired {
+			ridExpiredSub, err = ridRepo.DeleteExpiredSubscriptions(ctx, *locality, ridThreshold)
 			if err != nil {
-				return stacktrace.Propagate(err,
-					"Failed to list RID expired Subscriptions")
+				return fmt.Errorf("failed to delete expired RID subscriptions: %w", err)
 			}
-
-			if *deleteExpired {
-				for _, sub := range ridExpiredSub {
-					_, err := r.DeleteSubscription(ctx, sub)
-					if err != nil {
-						return stacktrace.Propagate(err,
-							"Failed to delete RID Subscription")
-					}
-				}
+		} else {
+			ridExpiredSub, err = ridRepo.ListExpiredSubscriptions(ctx, *locality, ridThreshold)
+			if err != nil {
+				return fmt.Errorf("failed to list RID expired subscriptions: %w", err)
 			}
-
 		}
-
-		return nil
-	}
-	if _, err = ridStore.Transact(ctx, dssstore.NewFuncOperation(ridAction)); err != nil {
-		return fmt.Errorf("failed to execute RID transaction: %w", err)
 	}
 
-	for _, opIntent := range expiredOpIntents {
-		logExpiredEntity("operational intent", opIntent.ID, scdThreshold, *deleteExpired, opIntent.EndTime != nil)
+	action := "found"
+	if *deleteExpired {
+		action = "deleted"
 	}
-	for _, sub := range scdExpiredSub {
-		logExpiredEntity("SCD subscription", sub.ID, scdThreshold, *deleteExpired, sub.EndTime != nil)
+	for _, id := range expiredOpIntents {
+		log.Printf("%s expired operational intent %s", action, id)
 	}
-	for _, isa := range expiredISAs {
-		logExpiredEntity("ISA", isa.ID, ridThreshold, *deleteExpired, isa.EndTime != nil)
+	for _, id := range scdExpiredSub {
+		log.Printf("%s expired SCD subscription %s", action, id)
 	}
-	for _, sub := range ridExpiredSub {
-		logExpiredEntity("RID subscription", sub.ID, ridThreshold, *deleteExpired, sub.EndTime != nil)
+	for _, id := range expiredISAs {
+		log.Printf("%s expired ISA %s", action, id)
+	}
+	for _, id := range ridExpiredSub {
+		log.Printf("%s expired RID subscription %s", action, id)
 	}
 	if len(expiredOpIntents) == 0 && len(scdExpiredSub) == 0 && len(expiredISAs) == 0 && len(ridExpiredSub) == 0 {
 		log.Printf("no SCD entity older than %s and no RID entity older than %s found", scdThreshold.String(), ridThreshold.String())
@@ -169,16 +154,3 @@ func evict(cmd *cobra.Command, _ []string) error {
 	}
 	return nil
 }
-
-func logExpiredEntity(entity string, entityID dssmodels.ID, threshold time.Time, deleted, hasEndTime bool) {
-	logMsg := "found"
-	if deleted {
-		logMsg = "deleted"
-	}
-
-	expMsg := "last update before %s (missing end time)"
-	if hasEndTime {
-		expMsg = "end time before %s"
-	}
-	log.Printf("%s %s %s; expired due to %s", logMsg, entity, entityID.String(), fmt.Sprintf(expMsg, threshold.String()))
-}

docs/operations/cleanup.md

@@ -38,9 +38,9 @@ The defaults shipped with the deployment tooling (`30m` TTL on RID running every
 
 ## Performance impact
 
-All expired entities are identified and removed within a single transaction. When the system is under heavy load, lock contention with concurrent transactions may cause the cleanup to fail. There is no risk of data inconsistency in this case - the cleanup may simply be retried.
+Each entity type (SCD operational intents, SCD subscriptions, RID ISAs, RID subscriptions) is evicted with a single statement. Entity types are deleted independently, so contention or failure on one does not affect the others. There is no risk of data inconsistency in this case - the cleanup may simply be retried.
 
-To mitigate this:
+To mitigate contention:
 
 - Run the cleanup during low-intensity periods (e.g. at night).
 - Clean up iteratively, starting with a lower TTL and progressively increasing it.

pkg/rid/operations/subscription_test.go

@@ -180,7 +180,11 @@ func (r *fakeSubscriptionRepo) MaxSubscriptionCountInCellsByOwner(ctx context.Co
 	return maxValue, nil
 }
 
-func (r *fakeSubscriptionRepo) ListExpiredSubscriptions(_ context.Context, _ string, _ time.Time) ([]*ridmodels.Subscription, error) {
+func (r *fakeSubscriptionRepo) ListExpiredSubscriptions(_ context.Context, _ string, _ time.Time) ([]dssmodels.ID, error) {
+	return nil, nil
+}
+
+func (r *fakeSubscriptionRepo) DeleteExpiredSubscriptions(_ context.Context, _ string, _ time.Time) ([]dssmodels.ID, error) {
 	return nil, nil
 }
 
@@ -229,7 +233,11 @@ func (r *fakeSubscriptionRepo) SearchISAs(_ context.Context, cells s2.CellUnion,
 	return isas, nil
 }
 
-func (r *fakeSubscriptionRepo) ListExpiredISAs(_ context.Context, _ string, _ time.Time) ([]*ridmodels.IdentificationServiceArea, error) {
+func (r *fakeSubscriptionRepo) ListExpiredISAs(_ context.Context, _ string, _ time.Time) ([]dssmodels.ID, error) {
+	panic("not implemented")
+}
+
+func (r *fakeSubscriptionRepo) DeleteExpiredISAs(_ context.Context, _ string, _ time.Time) ([]dssmodels.ID, error) {
 	panic("not implemented")
 }
 

pkg/rid/repos/isa.go

@@ -29,8 +29,11 @@ type ISA interface {
 	// SearchISAs returns all subscriptions ownded by "owner" in "cells".
 	SearchISAs(ctx context.Context, cells s2.CellUnion, earliest *time.Time, latest *time.Time) ([]*ridmodels.IdentificationServiceArea, error)
 
-	// ListExpiredISAs lists all expired ISAs based on writer
-	ListExpiredISAs(ctx context.Context, writer string, threshold time.Time) ([]*ridmodels.IdentificationServiceArea, error)
+	// ListExpiredISAs lists the IDs of all expired ISAs based on writer
+	ListExpiredISAs(ctx context.Context, writer string, threshold time.Time) ([]dssmodels.ID, error)
+
+	// DeleteExpiredISAs deletes all expired ISAs based on writer and returns the IDs of the deleted ISAs.
+	DeleteExpiredISAs(ctx context.Context, writer string, threshold time.Time) ([]dssmodels.ID, error)
 
 	// Count the number of existing ISA
 	CountISAs(ctx context.Context) (int64, error)

pkg/rid/repos/subscription.go

@@ -39,8 +39,11 @@ type Subscription interface {
 	// belonging to the given owner, and returns that number.
 	MaxSubscriptionCountInCellsByOwner(ctx context.Context, cells s2.CellUnion, owner dssmodels.Owner) (int, error)
 
-	// ListExpiredSubscriptions lists all expired Subscriptions based on writer.
-	ListExpiredSubscriptions(ctx context.Context, writer string, threshold time.Time) ([]*ridmodels.Subscription, error)
+	// ListExpiredSubscriptions lists the IDs of all expired Subscriptions based on writer.
+	ListExpiredSubscriptions(ctx context.Context, writer string, threshold time.Time) ([]dssmodels.ID, error)
+
+	// DeleteExpiredSubscriptions deletes all expired Subscriptions based on writer and returns the IDs of the deleted Subscriptions.
+	DeleteExpiredSubscriptions(ctx context.Context, writer string, threshold time.Time) ([]dssmodels.ID, error)
 
 	// Count the number of existing subscriptions
 	CountSubscriptions(ctx context.Context) (int64, error)

pkg/rid/store/memstore/identification_service_area.go

@@ -129,8 +129,14 @@ func (r *repo) SearchISAs(_ context.Context, cells s2.CellUnion, earliest *time.
 	return out, nil
 }
 
-func (r *repo) ListExpiredISAs(_ context.Context, writer string, threshold time.Time) ([]*ridmodels.IdentificationServiceArea, error) {
-	return listExpired[ridmodels.IdentificationServiceArea](r.state.ISAs, writer, threshold, dssmodels.MaxResultLimit), nil
+// TODO: Implement when raftstore evict is implemented (#1718)
+func (r *repo) ListExpiredISAs(_ context.Context, writer string, threshold time.Time) ([]dssmodels.ID, error) {
+	return listExpired(r.state.ISAs, writer, threshold, dssmodels.MaxResultLimit), nil
+}
+
+// TODO: Implement when raftstore evict is implemented (#1718)
+func (r *repo) DeleteExpiredISAs(_ context.Context, writer string, threshold time.Time) ([]dssmodels.ID, error) {
+	return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "DeleteExpiredISAs not implemented for memstore")
 }
 
 func (r *repo) CountISAs(_ context.Context) (int64, error) {

pkg/rid/store/memstore/store.go

@@ -139,10 +139,9 @@ func findForWrite[R versionedRecord](store map[dssmodels.ID]R, id dssmodels.ID,
 	return rec, true
 }
 
-type expiringRecord[M any] interface {
+type expiringRecord interface {
 	endTime() *time.Time
 	writerName() string
-	toModel() *M
 }
 
 func (rec *isaRecord) endTime() *time.Time { return rec.EndTime }
@@ -153,18 +152,18 @@ func (rec *subscriptionRecord) endTime() *time.Time { return rec.EndTime }
 
 func (rec *subscriptionRecord) writerName() string { return rec.Writer }
 
-// listExpired returns the records whose end time is at or before threshold and
+// listExpired returns the IDs of the records whose end time is at or before threshold and
 // whose writer matches. A limit of 0 means unlimited.
-func listExpired[M any, R expiringRecord[M]](store map[dssmodels.ID]R, writer string, threshold time.Time, limit int) []*M {
-	var out []*M
-	for _, rec := range store {
+func listExpired[R expiringRecord](store map[dssmodels.ID]R, writer string, threshold time.Time, limit int) []dssmodels.ID {
+	var out []dssmodels.ID
+	for id, rec := range store {
 		if t := rec.endTime(); t == nil || t.After(threshold) { // TODO: Don't allow endtime to be null, see #1492
 			continue
 		}
 		if rec.writerName() != writer {
 			continue
 		}
-		out = append(out, rec.toModel())
+		out = append(out, id)
 
 		if limit > 0 && len(out) > limit { // This mimics sqlstore behaviour, but it's not very good.
 			break

pkg/rid/store/memstore/subscriptions.go

@@ -185,9 +185,15 @@ func (r *repo) MaxSubscriptionCountInCellsByOwner(ctx context.Context, cells s2.
 	return slices.Max(slices.Collect(maps.Values(counts))), nil
 }
 
-func (r *repo) ListExpiredSubscriptions(_ context.Context, writer string, threshold time.Time) ([]*ridmodels.Subscription, error) {
+// TODO: Implement when raftstore evict is implemented (#1718)
+func (r *repo) ListExpiredSubscriptions(_ context.Context, writer string, threshold time.Time) ([]dssmodels.ID, error) {
 	// TODO: This mimics sqlstore inconsistency of not limiting results there, compared to ISAs. Should it be normalized?
-	return listExpired[ridmodels.Subscription](r.state.Subscriptions, writer, threshold, 0), nil
+	return listExpired(r.state.Subscriptions, writer, threshold, 0), nil
+}
+
+// TODO: Implement when raftstore evict is implemented (#1718)
+func (r *repo) DeleteExpiredSubscriptions(_ context.Context, writer string, threshold time.Time) ([]dssmodels.ID, error) {
+	return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "DeleteExpiredSubscriptions not implemented for memstore")
 }
 
 func (r *repo) CountSubscriptions(_ context.Context) (int64, error) {

pkg/rid/store/raftstore/identification_service_area.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/golang/geo/s2"
+	dsserr "github.com/interuss/dss/pkg/errors"
 	dssmodels "github.com/interuss/dss/pkg/models"
 	"github.com/interuss/dss/pkg/raftstore/consensus"
 	ridmodels "github.com/interuss/dss/pkg/rid/models"
@@ -18,7 +19,7 @@ const (
 	insertISA       consensus.RequestType[*ridmodels.IdentificationServiceArea]   = "insertISA"
 	updateISA       consensus.RequestType[*ridmodels.IdentificationServiceArea]   = "updateISA"
 	searchISAs      consensus.RequestType[[]*ridmodels.IdentificationServiceArea] = "searchISAs"
-	listExpiredISAs consensus.RequestType[[]*ridmodels.IdentificationServiceArea] = "listExpiredISAs"
+	listExpiredISAs consensus.RequestType[[]dssmodels.ID]                         = "listExpiredISAs"
 	countISAs       consensus.RequestType[int64]                                  = "countISAs"
 )
 
@@ -69,7 +70,7 @@ func (r *repo) SearchISAs(ctx context.Context, cells s2.CellUnion, earliest *tim
 	return r.consensus.HandleReadRequest(ctx, searchISAs, buf)
 }
 
-func (r *repo) ListExpiredISAs(ctx context.Context, writer string, threshold time.Time) ([]*ridmodels.IdentificationServiceArea, error) {
+func (r *repo) ListExpiredISAs(ctx context.Context, writer string, threshold time.Time) ([]dssmodels.ID, error) {
 	buf, err := json.Marshal(expiredPayload{Writer: writer, Threshold: threshold})
 	if err != nil {
 		return nil, stacktrace.Propagate(err, "failed to marshal payload")
@@ -78,6 +79,10 @@ func (r *repo) ListExpiredISAs(ctx context.Context, writer string, threshold tim
 	return r.consensus.HandleReadRequest(ctx, listExpiredISAs, buf)
 }
 
+func (r *repo) DeleteExpiredISAs(_ context.Context, writer string, threshold time.Time) ([]dssmodels.ID, error) {
+	return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "DeleteExpiredISAs not implemented for raftstore")
+}
+
 func (r *repo) CountISAs(ctx context.Context) (int64, error) {
 	return r.consensus.HandleReadRequest(ctx, countISAs, nil)
 }