Fix ss leak main

pkg/cnservice/distributed_tae.go

@@ -100,6 +100,9 @@ func (s *service) initDistributedTAE(
 		return err
 	}
 
+	// Start unified GC scheduler
+	go cnEngine.RunGCScheduler(ctx)
+
 	ss, ok := runtime.ServiceRuntime(s.cfg.UUID).GetGlobalVariables(runtime.StatusServer)
 	if ok {
 		statusServer := ss.(*status.Server)

pkg/vm/engine/disttae/db.go

@@ -18,9 +18,10 @@ import (
 	"context"
 	"strconv"
 	"strings"
-	"sync"
 	"time"
 
+	"go.uber.org/zap"
+
 	"github.com/matrixorigin/matrixone/pkg/catalog"
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/mpool"
@@ -487,24 +488,9 @@ func (e *Engine) getOrCreateSnapPartBy(
 			tbl.db.op.Txn().DebugString())
 	}
 
-	//check whether the snapshot partitions are available for reuse.
-	e.mu.Lock()
-	tblSnaps, ok := e.mu.snapParts[[2]uint64{tbl.db.databaseId, tbl.tableId}]
-	if !ok {
-		e.mu.snapParts[[2]uint64{tbl.db.databaseId, tbl.tableId}] = &struct {
-			sync.Mutex
-			snaps []*logtailreplay.Partition
-		}{}
-		tblSnaps = e.mu.snapParts[[2]uint64{tbl.db.databaseId, tbl.tableId}]
-	}
-	e.mu.Unlock()
-
-	tblSnaps.Lock()
-	defer tblSnaps.Unlock()
-	for _, snap := range tblSnaps.snaps {
-		if snap.Snapshot().CanServe(ts) {
-			return snap.Snapshot(), nil
-		}
+	// Try to find existing snapshot that can serve this timestamp
+	if ps := e.snapshotMgr.Find(tbl.db.databaseId, tbl.tableId, ts); ps != nil {
+		return ps, nil
 	}
 
 	//new snapshot partition and apply checkpoints into it.
@@ -555,14 +541,35 @@ func (e *Engine) getOrCreateSnapPartBy(
 		return nil
 	})
 	if err != nil {
-		logutil.Infof("Snapshot consumeSnapCkps failed, err:%v", err)
+		logutil.Error("Snapshot consumeSnapCkps failed", zap.Error(err))
 		return nil, err
 	}
-	if snap.Snapshot().CanServe(ts) {
-		tblSnaps.snaps = append(tblSnaps.snaps, snap)
-		return snap.Snapshot(), nil
+	if !snap.Snapshot().CanServe(ts) {
+		return nil, moerr.NewInternalErrorNoCtxf(
+			"snapshot partition cannot serve ts after consuming checkpoints, ts:%s, table:%s",
+			ts.ToString(), tbl.tableName)
 	}
-	panic("impossible path")
+
+	// Add the new snapshot with LRU eviction
+	ps := e.snapshotMgr.Add(
+		tbl.db.databaseId,
+		tbl.tableId,
+		snap,
+		tbl.tableName,
+		ts,
+	)
+
+	// Log total snapshots
+	metrics := e.snapshotMgr.GetMetrics()
+	logutil.Info(
+		"Snapshot-Added",
+		zap.Int64("total-snaps-global", metrics.TotalSnapshots.Load()),
+	)
+
+	// Trigger GC check if needed
+	e.snapshotMgr.MaybeStartGC()
+
+	return ps, nil
 }
 
 func (e *Engine) GetOrCreateLatestPart(

pkg/vm/engine/disttae/engine.go

@@ -106,10 +106,9 @@ func New(
 			},
 		),
 	}
-	e.mu.snapParts = make(map[[2]uint64]*struct {
-		sync.Mutex
-		snaps []*logtailreplay.Partition
-	})
+	// Initialize snapshot manager
+	e.snapshotMgr = NewSnapshotManager()
+	e.snapshotMgr.Init()
 
 	pool, err := ants.NewPool(GCPoolSize)
 	if err != nil {
@@ -850,3 +849,46 @@ func (e *Engine) LatestLogtailAppliedTime() timestamp.Timestamp {
 func (e *Engine) HasTempEngine() bool {
 	return false
 }
+
+// RunGCScheduler runs all GC tasks in a single goroutine with different intervals
+func (e *Engine) RunGCScheduler(ctx context.Context) {
+	unusedTableTicker := time.NewTicker(unsubscribeProcessTicker)
+	partitionStateTicker := time.NewTicker(gcPartitionStateTicker)
+	snapshotTicker := time.NewTicker(gcSnapshotTicker)
+
+	defer unusedTableTicker.Stop()
+	defer partitionStateTicker.Stop()
+	defer snapshotTicker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			logutil.Infof("GC scheduler exit.")
+			return
+
+		case <-unusedTableTicker.C:
+			// GC unused tables in PushClient
+			e.pClient.TryGC(ctx)
+
+		case <-partitionStateTicker.C:
+			// GC partition states
+			e.gcPartitionState(ctx)
+
+		case <-snapshotTicker.C:
+			// GC snapshot partitions
+			e.snapshotMgr.MaybeStartGC()
+		}
+	}
+}
+
+// gcPartitionState runs GC for partition states
+func (e *Engine) gcPartitionState(ctx context.Context) {
+	if e.pClient.subscriber == nil {
+		return
+	}
+	if !e.pClient.receivedLogTailTime.ready.Load() {
+		return
+	}
+	logutil.Debugf("Running partition state GC")
+	e.pClient.doGCPartitionState(ctx, e)
+}

pkg/vm/engine/disttae/logtail_consumer.go

@@ -103,6 +103,7 @@ const (
 var (
 	unsubscribeProcessTicker = 20 * time.Minute
 	gcPartitionStateTicker   = 20 * time.Minute
+	gcSnapshotTicker         = 20 * time.Minute // snapshot GC interval
 
 	defaultGetLogTailAddrTimeoutDuration = time.Minute * 10
 	defaultServerTimeout                 = time.Minute * 10
@@ -1077,6 +1078,18 @@ func (c *PushClient) partitionStateGCTicker(ctx context.Context, e *Engine) {
 	}
 }
 
+// TryGC checks if ready and runs GC for PushClient resources (unused tables)
+func (c *PushClient) TryGC(ctx context.Context) {
+	if c.subscriber == nil {
+		return
+	}
+	if !c.subscriber.ready() {
+		return
+	}
+	logutil.Debugf("%s Running unused table GC", logTag)
+	c.doGCUnusedTable(ctx)
+}
+
 // subscribedTable used to record table subscribed status.
 // only if m[table T] = true, T has been subscribed.
 type subscribedTable struct {
@@ -1701,8 +1714,6 @@ func (e *Engine) InitLogTailPushModel(ctx context.Context, timestampWaiter clien
 	// Start a goroutine that never stops to receive logtail from TN logtail server.
 	go e.pClient.run(ctx, e)
 
-	go e.pClient.unusedTableGCTicker(ctx)
-	go e.pClient.partitionStateGCTicker(ctx, e)
 	return nil
 }