feat(server): prune sessions past a configurable retention window

.env.enterprise

@@ -55,3 +55,7 @@ SHELLHUB_OBJECT_STORAGE_SECRET_KEY=password
 # NOTICE: Leave empty for license-less development; the API then starts
 # without a license and gated features stay disabled.
 SHELLHUB_LICENSE_FILE=
+
+# How long sessions and their events are kept after the session started (days).
+# 0 keeps them indefinitely.
+SHELLHUB_SESSION_RETENTION_DAYS=180

docker-compose.enterprise.yml

@@ -27,6 +27,7 @@ services:
       # GeoIP (MaxMind) database source, read by the enterprise binary
       - MAXMIND_MIRROR=${SHELLHUB_MAXMIND_MIRROR:-}
       - MAXMIND_LICENSE=${SHELLHUB_MAXMIND_LICENSE:-}
+      - SHELLHUB_SESSION_RETENTION_DAYS=${SHELLHUB_SESSION_RETENTION_DAYS}
     secrets:
       - api_private_key
       - api_public_key

server/api/services/service.go

@@ -25,6 +25,7 @@ type service struct {
 	billing           BillingProvider
 	licenseEvaluator  LicenseEvaluator
 	firewallEvaluator FirewallEvaluator
+	recordingPruner   SessionRecordingPruner
 }
 
 type Service interface {
@@ -94,6 +95,12 @@ func WithFirewallEvaluator(fe FirewallEvaluator) Option {
 	}
 }
 
+func WithSessionRecordingPruner(rp SessionRecordingPruner) Option {
+	return func(service *APIService) {
+		service.recordingPruner = rp
+	}
+}
+
 func NewService(store store.Store, privKey *rsa.PrivateKey, pubKey *rsa.PublicKey, cache cache.Cache, options ...Option) *APIService {
 	if privKey == nil || pubKey == nil {
 		var err error
@@ -114,6 +121,7 @@ func NewService(store store.Store, privKey *rsa.PrivateKey, pubKey *rsa.PublicKe
 			billing:           nil, // injected via WithBilling option
 			licenseEvaluator:  nil, // injected via WithLicenseEvaluator option
 			firewallEvaluator: nil, // injected via WithFirewallEvaluator option
+			recordingPruner:   nil, // injected via WithSessionRecordingPruner option
 		},
 	}
 

server/api/services/session-recording.go

@@ -0,0 +1,95 @@
+package services
+
+import (
+	"context"
+
+	"github.com/shellhub-io/shellhub/pkg/cache"
+	"github.com/shellhub-io/shellhub/server/api/store"
+)
+
+// SessionRecordingPrunerFactoryFunc constructs a SessionRecordingPruner from the core store and
+// cache. Enterprise packages register a factory via RegisterSessionRecordingPruner in their
+// init() functions; it runs during server setup.
+type SessionRecordingPrunerFactoryFunc func(ctx context.Context, store store.Store, cache cache.Cache) (SessionRecordingPruner, error)
+
+var sessionRecordingPrunerFactory SessionRecordingPrunerFactoryFunc
+
+// RegisterSessionRecordingPruner registers the factory that creates the recording pruner.
+// It must be called before the server's Setup() runs.
+func RegisterSessionRecordingPruner(f SessionRecordingPrunerFactoryFunc) {
+	sessionRecordingPrunerFactory = f
+}
+
+// SessionRecordingPrunerFactory returns the registered factory, or nil in Community Edition
+// builds.
+func SessionRecordingPrunerFactory() SessionRecordingPrunerFactoryFunc {
+	return sessionRecordingPrunerFactory
+}
+
+// SessionRecordingPruner discards the stored recordings of sessions that retention is about to
+// delete.
+//
+// A recording is an object, not a row, and nothing in the schema points at it: it is found by
+// composing a key from the session's UID. Deleting the session row therefore does not delete the
+// recording — it destroys the only thing that could still name it. Everything else in this seam
+// follows from that asymmetry.
+type SessionRecordingPruner interface {
+	// DeleteRecordings removes the recordings of the given sessions, whatever seats they had,
+	// and returns the subset it managed to purge.
+	//
+	// Returning a subset rather than failing the batch is what keeps one unreachable object from
+	// halting retention: the caller deletes the rows it names and leaves the rest, so a session
+	// whose recording cannot be removed holds up nothing but itself. The error is reserved for a
+	// failure that makes the whole batch moot, such as a cancelled context.
+	DeleteRecordings(ctx context.Context, uids []string) ([]string, error)
+}
+
+// pruneRecordings discards the recordings of the recorded sessions in the batch and returns the
+// sessions whose rows may now be deleted.
+//
+// Without a pruner — Community Edition, or an enterprise instance with no object storage — no
+// session owns anything outside the database, so the whole batch is deletable as it stands.
+func (s *service) pruneRecordings(ctx context.Context, sessions []store.ExpiredSession) ([]string, error) {
+	uids := make([]string, 0, len(sessions))
+	recorded := make([]string, 0, len(sessions))
+
+	for _, session := range sessions {
+		uids = append(uids, session.UID)
+
+		if session.Recorded {
+			recorded = append(recorded, session.UID)
+		}
+	}
+
+	if s.recordingPruner == nil || len(recorded) == 0 {
+		return uids, nil
+	}
+
+	purged, err := s.recordingPruner.DeleteRecordings(ctx, recorded)
+	if err != nil {
+		return nil, err
+	}
+
+	// Everything that was never recorded, plus the recordings actually purged. A session left
+	// out here keeps its row, so its object stays reachable for the next run to retry.
+	deletable := make([]string, 0, len(uids))
+	purgedSet := make(map[string]struct{}, len(purged))
+
+	for _, uid := range purged {
+		purgedSet[uid] = struct{}{}
+	}
+
+	for _, session := range sessions {
+		if !session.Recorded {
+			deletable = append(deletable, session.UID)
+
+			continue
+		}
+
+		if _, ok := purgedSet[session.UID]; ok {
+			deletable = append(deletable, session.UID)
+		}
+	}
+
+	return deletable, nil
+}

server/api/services/task.go

@@ -21,6 +21,20 @@ const (
 	CronEphemeralCleanup          = worker.CronSpec("*/5 * * * *")
 	CronEnrollmentCallbackCleanup = worker.CronSpec("0 4 * * *")
 	CronSSHApprovalCleanup        = worker.CronSpec("*/10 * * * *")
+	CronSessionCleanup            = worker.CronSpec("0 1 * * *")
+)
+
+const (
+	// A session cascades into its events, so one batch is already thousands of rows.
+	sessionCleanupBatchSize = 1000
+
+	// Together with the batch size this caps a run at 100k sessions. An instance adopting
+	// retention for the first time can have years to shed, and draining all of it in one night
+	// is the write storm the batching exists to avoid.
+	sessionCleanupMaxBatches = 100
+
+	// Leaves room between batches for live traffic and for autovacuum to follow behind.
+	sessionCleanupBatchPause = 200 * time.Millisecond
 )
 
 func (s *service) DeviceCleanup() worker.CronHandler {
@@ -57,6 +71,93 @@ func (s *service) SSHApprovalCleanup() worker.CronHandler {
 	}
 }
 
+// SessionCleanup enforces the instance's session retention window: sessions that started longer
+// ago than retention are deleted, taking their events and recordings with them.
+//
+// A retention that is not positive means "keep forever" and prunes nothing. The guard matters
+// more than it looks: read as a window, a zero would put the cutoff at now and delete every
+// session on the instance.
+func (s *service) SessionCleanup(retention time.Duration) worker.CronHandler {
+	return func(ctx context.Context) error {
+		return s.sessionCleanup(ctx, retention, sessionCleanupBatchPause)
+	}
+}
+
+// sessionCleanup takes the pause as an argument so a test can drive the batching loop without
+// waiting it out.
+func (s *service) sessionCleanup(ctx context.Context, retention, pause time.Duration) error {
+	if retention <= 0 {
+		return nil
+	}
+
+	cutoff := clock.Now().Add(-retention)
+
+	total := int64(0)
+	batches := 0
+
+	for batches < sessionCleanupMaxBatches {
+		sessions, err := s.store.SessionListExpired(ctx, cutoff, sessionCleanupBatchSize)
+		if err != nil {
+			log.WithError(err).WithField("deleted", total).Error("failed to list expired sessions")
+
+			return err
+		}
+
+		if len(sessions) == 0 {
+			break
+		}
+
+		// Recordings first, rows second, because the row is the only thing that can still name
+		// the object. Sessions whose recording could not be purged are left out and keep their
+		// rows, so the next run finds them again.
+		deletable, err := s.pruneRecordings(ctx, sessions)
+		if err != nil {
+			log.WithError(err).WithField("deleted", total).Error("failed to prune recordings of expired sessions")
+
+			return err
+		}
+
+		// The batch is not empty but nothing in it can be deleted, so every session in it is
+		// blocked on its recording. Retrying inside this run would list the same rows again and
+		// spin until the cap; leave it for the next one, by which time the storage may answer.
+		if len(deletable) == 0 {
+			log.WithFields(log.Fields{"deleted": total, "blocked": len(sessions)}).
+				Warn("no expired session in the batch could be deleted; ending the run")
+
+			break
+		}
+
+		deleted, err := s.store.SessionDeleteMany(ctx, deletable)
+		if err != nil {
+			log.WithError(err).WithField("deleted", total).Error("failed to prune expired sessions")
+
+			return err
+		}
+
+		total += deleted
+		batches++
+
+		// A batch that came back short means the store ran out of sessions older than the
+		// cutoff, so there is nothing left for this run to do.
+		if len(sessions) < sessionCleanupBatchSize {
+			break
+		}
+
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(pause):
+		}
+	}
+
+	if total > 0 {
+		log.WithFields(log.Fields{"deleted": total, "cutoff": cutoff, "capped": batches == sessionCleanupMaxBatches}).
+			Info("pruned sessions past the retention window")
+	}
+
+	return nil
+}
+
 // EnrollmentCallbackCleanup prunes single-use callback redemption records once older than the maximum
 // token TTL, past which the token has expired and can no longer gate a replay. The table only gains a
 // row per resolved deferred webhook, so this keeps its growth bounded.