fix(repository): add repository clean by instance instead of globally

engine/repositories/cleaner_test.go

@@ -0,0 +1,68 @@
+package repositories
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/ovh/cds/sdk"
+)
+
+func Test_vacuumFileSystemCleanerFunc(t *testing.T) {
+	s, err := newTestService(t)
+	require.NoError(t, err)
+	s.Cfg.Basedir = t.TempDir()
+
+	otherDao := dao{store: s.Cache, hostname: "other-instance"}
+
+	tests := []struct {
+		name       string
+		protect    func(repoID string)
+		expectKept bool
+	}{
+		{
+			name:       "clone without last access is removed",
+			protect:    nil,
+			expectKept: false,
+		},
+		{
+			name:       "clone recently used by this instance is kept",
+			protect:    func(repoID string) { s.dao.saveLastAccess(repoID, time.Now().Add(time.Minute), 60) },
+			expectKept: true,
+		},
+		{
+			name:       "last access from another instance does not protect the clone",
+			protect:    func(repoID string) { otherDao.saveLastAccess(repoID, time.Now().Add(time.Minute), 60) },
+			expectKept: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			repoID := sdk.UUID()
+			path := filepath.Join(s.Cfg.Basedir, repoID)
+			require.NoError(t, os.MkdirAll(path, os.FileMode(0700)))
+			t.Cleanup(func() {
+				_ = s.Cache.Delete(s.dao.lastAccessKey(repoID))
+				_ = s.Cache.Delete(otherDao.lastAccessKey(repoID))
+			})
+
+			if tt.protect != nil {
+				tt.protect(repoID)
+			}
+
+			require.NoError(t, s.vacuumFileSystemCleanerFunc(context.TODO(), repoID))
+
+			_, err := os.Stat(path)
+			if tt.expectKept {
+				require.NoError(t, err, "clone directory must still exist")
+			} else {
+				require.True(t, os.IsNotExist(err), "clone directory must have been removed")
+			}
+		})
+	}
+}

engine/repositories/dao.go

@@ -12,14 +12,25 @@ import (
 )
 
 var (
-	rootKey       = cache.Key("repositories", "operations")
-	processorKey  = cache.Key("repositories", "processor")
-	locksKey      = cache.Key("repositories", "locks")
-	lastAccessKey = cache.Key("repositories", "lastAccess")
+	rootKey           = cache.Key("repositories", "operations")
+	processorKey      = cache.Key("repositories", "processor")
+	locksKey          = cache.Key("repositories", "locks")
+	lastAccessRootKey = cache.Key("repositories", "lastAccess")
 )
 
 type dao struct {
 	store cache.Store
+	// hostname scopes lastAccess keys to this instance: clones live on a local
+	// filesystem, so retention must reflect each instance's own usage.
+	hostname string
+}
+
+func (d *dao) lastAccessKey(repoUUID string) string {
+	return cache.Key(lastAccessRootKey, d.hostname, repoUUID)
+}
+
+func (d *dao) saveLastAccess(repoUUID string, expiration time.Time, ttlSeconds int) {
+	d.store.SetWithTTL(d.lastAccessKey(repoUUID), expiration, ttlSeconds)
 }
 
 func (d *dao) saveOperation(o *sdk.Operation) error {
@@ -88,7 +99,7 @@ func (d *dao) unlock(ctx context.Context, uuid string) error {
 }
 
 func (d *dao) isExpired(ctx context.Context, uuid string) (time.Time, bool) {
-	k := cache.Key(lastAccessKey, uuid)
+	k := d.lastAccessKey(uuid)
 	var v time.Time
 	find, err := d.store.Get(k, &v)
 	if err != nil {

engine/repositories/dao_test.go

@@ -0,0 +1,32 @@
+package repositories
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/ovh/cds/sdk"
+)
+
+func Test_daoLastAccessPerInstance(t *testing.T) {
+	s, err := newTestService(t)
+	require.NoError(t, err)
+
+	daoA := dao{store: s.Cache, hostname: "instance-a"}
+	daoB := dao{store: s.Cache, hostname: "instance-b"}
+
+	repoID := sdk.UUID()
+	t.Cleanup(func() {
+		_ = s.Cache.Delete(daoA.lastAccessKey(repoID))
+	})
+
+	daoA.saveLastAccess(repoID, time.Now().Add(time.Minute), 60)
+
+	_, expired := daoA.isExpired(context.TODO(), repoID)
+	require.False(t, expired, "repository must not be expired for the instance that accessed it")
+
+	_, expired = daoB.isExpired(context.TODO(), repoID)
+	require.True(t, expired, "a last access from another instance must not protect this instance's clone")
+}

engine/repositories/processor.go

@@ -7,7 +7,6 @@ import (
 
 	"github.com/rockbears/log"
 
-	"github.com/ovh/cds/engine/cache"
 	"github.com/ovh/cds/sdk"
 	cdslog "github.com/ovh/cds/sdk/log"
 )
@@ -76,7 +75,7 @@ func (s *Service) do(ctx context.Context, op sdk.Operation) error {
 		ttl := int(ttlDuration.Seconds())
 		ttlTime := time.Now().Add(ttlDuration)
 		log.Info(ctx, "%s protected until %s", r.ID(), ttlTime.String())
-		s.dao.store.SetWithTTL(cache.Key(lastAccessKey, r.ID()), ttlTime, ttl)
+		s.dao.saveLastAccess(r.ID(), ttlTime, ttl)
 	}()
 
 	switch {

engine/repositories/repositories.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"os"
 	"time"
 
 	"github.com/gorilla/mux"
@@ -114,9 +115,15 @@ func (s *Service) Serve(c context.Context) error {
 		MaxHeaderBytes: 1 << 20,
 	}
 
+	hostname, err := os.Hostname()
+	if err != nil {
+		return sdk.WrapError(err, "unable to get hostname")
+	}
+
 	//Set the dao
 	s.dao = dao{
-		store: s.Cache,
+		store:    s.Cache,
+		hostname: hostname,
 	}
 
 	log.Info(ctx, "Initializing processor...")

engine/repositories/repositories_test.go

@@ -72,7 +72,8 @@ func newTestService(t *testing.T) (*Service, error) {
 	}
 
 	service.dao = dao{
-		store: service.Cache,
+		store:    service.Cache,
+		hostname: "test-instance",
 	}
 
 	return service, nil