diff --git a/internal/application/service/tenant_skill_bundle.go b/internal/application/service/tenant_skill_bundle.go index fd23bf3a27..5f5bc18b6f 100644 --- a/internal/application/service/tenant_skill_bundle.go +++ b/internal/application/service/tenant_skill_bundle.go @@ -36,7 +36,8 @@ type SkillBundle struct { Description string Instructions string // SHA256 is over the uploaded bytes, so re-uploading the same archive is - // recognisable in the UI and in the ledger. + // recognisable in the UI and in the ledger, and a ready skill with this + // digest can skip a billed snapshot rebuild. SHA256 string // Files maps skill-root-relative paths to contents, SKILL.md included. Files map[string][]byte diff --git a/internal/application/service/tenant_skill_install.go b/internal/application/service/tenant_skill_install.go index 0644016dd8..83b24929f5 100644 --- a/internal/application/service/tenant_skill_install.go +++ b/internal/application/service/tenant_skill_install.go @@ -8,6 +8,7 @@ import ( "path" "sort" "strings" + "sync" "time" "github.com/google/uuid" @@ -63,6 +64,12 @@ func (s *TenantSkillService) InstallSkill( if err != nil { return "", err } + if s.canSkipInstall(ctx, existing, bundle) { + if err := s.refreshSkippedBundle(ctx, existing, archive); err != nil { + return "", fmt.Errorf("store bundle for skill %s: %w", existing.ID, err) + } + return existing.ID, nil + } skillID := uuid.NewString() now := s.now() @@ -201,6 +208,14 @@ func (s *TenantSkillService) runInstall( return nil } + // From here on this run is the row's owner, and everything below can take + // minutes. The heartbeat is what tells a second upload of the same archive + // (and the reaper) that those minutes are work rather than a dead process. + // It is deferred before it is stopped explicitly below, so a failure path + // still stops it ahead of the deferred failSkill. + stopHeartbeat := s.startInstallHeartbeat(ctx, tenantID, configID, skillID) + defer stopHeartbeat() + // The name comes from SKILL.md and is already validated on parse, so a // rejection here means the bundle was accepted by a looser rule than the // one the image path enforces. Failing before any sandbox work keeps that @@ -349,6 +364,10 @@ func (s *TenantSkillService) runInstall( return err } pointerSwitched = true + // The heartbeat writes the whole row, so it must be gone before the + // terminal "ready" write below: a beat landing after it would put the row + // back to installing and have the reaper fail a skill that is serving. + stopHeartbeat() s.markPreviousSnapshotsSuperseded(ctx, tenantID, configID, installRowID) // The terminal write is the one that must not be best-effort: the pointer @@ -873,9 +892,10 @@ func (s *TenantSkillService) failSkill( // installStillOwnsTheRow is the lock-side counterpart of InstallSkill's // optimistic row write. A remove that ran first deleted the row; a newer // upload of the same name replaced BundleSHA256; a queued remove flipped the -// status. Any of those means this run must not snapshot — failSkill would -// stamp the newer owner's row, and a snapshot with no matching row is an -// orphan the ledger cannot name. +// status; a sibling retry of the same archive found the first run had already +// landed in the live image. Any of those means this run must not snapshot — +// failSkill would stamp the newer owner's row, and a snapshot with no matching +// row is an orphan the ledger cannot name. func (s *TenantSkillService) installStillOwnsTheRow( ctx context.Context, tenantID uint64, configID, skillID string, bundle *SkillBundle, ) (bool, error) { @@ -892,9 +912,147 @@ func (s *TenantSkillService) installStillOwnsTheRow( if bundle != nil && current.BundleSHA256 != "" && current.BundleSHA256 != bundle.SHA256 { return false, nil } + if current.Status == types.SkillStatusReady { + _, inImage, ok := s.skillFilesInLiveImage(ctx, current) + if ok && inImage { + return false, nil + } + } return true, nil } +// canSkipInstall reports whether this upload is a no-op. Re-uploading the +// exact archive of a skill that is already ready (and still in the live image) +// must not boot a billed sandbox or grow a new snapshot. An install of the +// same bytes that is still beating is the same situation: the first run owns +// the work. +// +// Only a ready row is answered from the image. An installing row is answered +// from the heartbeat alone, deliberately: the ledger records which skill an +// install snapshotted, not which archive, so a row that is installing bundle +// B while the image still carries the earlier bundle A would look "already +// installed" and this upload would report a success that never happened. +// +// A failed skill with the same digest is a retry: the previous attempt never +// made it into the image. A removal in flight is not a skip either — taking +// the row back to installing is how an upload cancels it. +func (s *TenantSkillService) canSkipInstall( + ctx context.Context, existing *types.TenantSkillEntity, bundle *SkillBundle, +) bool { + if existing == nil || bundle == nil { + return false + } + if existing.BundleSHA256 == "" || existing.BundleSHA256 != bundle.SHA256 { + return false + } + switch existing.Status { + case types.SkillStatusInstalling: + return s.installIsInFlight(existing) + case types.SkillStatusReady: + _, inImage, ok := s.skillFilesInLiveImage(ctx, existing) + return ok && inImage + default: + return false + } +} + +// installIsInFlight reports whether an installing row still belongs to a live +// process. The answer is the heartbeat: a running install restamps +// InstallingSince every skillInstallHeartbeatInterval, so silence past +// skillInstallInFlightSkip means the process is gone and the next upload must +// be allowed to start a new run rather than wait for the stuck-run reaper. +// +// Reading the submission time instead would force a choice between calling a +// slow install dead — a single agent command may take installCommandTimeout, +// and an install runs several — and leaving a dead one unrecoverable. +func (s *TenantSkillService) installIsInFlight(existing *types.TenantSkillEntity) bool { + if existing == nil || existing.InstallingSince == nil { + return false + } + return !existing.InstallingSince.Before(s.clock()().Add(-skillInstallInFlightSkip)) +} + +// startInstallHeartbeat keeps this run's liveness visible while it works, and +// returns the stop function that must be called before any terminal write. +// +// The heartbeat writes the whole row, so it would otherwise race the "ready" +// write past the pointer switch and revive an installing status. Both callers +// stop it before that point: runInstall stops it the moment the pointer moves, +// and the deferred stop runs before the deferred failSkill. +func (s *TenantSkillService) startInstallHeartbeat( + ctx context.Context, tenantID uint64, configID, skillID string, +) func() { + interval := s.installHeartbeat + if interval <= 0 { + interval = skillInstallHeartbeatInterval + } + beatCtx, stop := context.WithCancel(ctx) + done := make(chan struct{}) + go func() { + defer close(done) + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-beatCtx.Done(): + return + case <-ticker.C: + s.beatInstallHeartbeat(beatCtx, tenantID, configID, skillID) + } + } + }() + var once sync.Once + return func() { + once.Do(func() { + stop() + <-done + }) + } +} + +// beatInstallHeartbeat stamps InstallingSince for a row this run still owns. +// A row that has left the installing status belongs to a newer upload, a +// queued removal, or a finished run, and reviving its timestamp would hide +// one of those from the reaper. +func (s *TenantSkillService) beatInstallHeartbeat( + ctx context.Context, tenantID uint64, configID, skillID string, +) { + current, err := s.skills.GetSkill(ctx, tenantID, configID, skillID) + if err != nil { + logger.Warnf(ctx, "[skill] load %s for install heartbeat failed: %v", skillID, err) + return + } + if current == nil || current.Status != types.SkillStatusInstalling { + return + } + at := s.clock()() + current.InstallingSince = &at + if err := s.skills.UpdateSkill(ctx, current); err != nil { + logger.Warnf(ctx, "[skill] install heartbeat for %s failed: %v", skillID, err) + } +} + +// refreshSkippedBundle stores the uploaded archive even when the image work +// is skipped. read_skill serves file contents from it, so a re-upload of a +// ready skill is how a missing object-store blob gets repaired without +// growing a new snapshot. A failure here is returned to the caller rather +// than turning the ready row into a failed install. +func (s *TenantSkillService) refreshSkippedBundle( + ctx context.Context, existing *types.TenantSkillEntity, archive []byte, +) error { + if existing == nil { + return nil + } + ref, err := s.saveBundle(ctx, existing.TenantID, existing.ID, archive) + if err != nil { + return err + } + return s.updateSkillFields(ctx, existing.TenantID, existing.SandboxConfigID, existing.ID, + func(e *types.TenantSkillEntity) { + e.BundleRef = ref + }) +} + // startMaintenanceSession opens the session one image operation runs in. The // operation name is carried into the session because the transcript is kept // deliberately, for troubleshooting: filing a removal's under "Skill install" diff --git a/internal/application/service/tenant_skill_install_test.go b/internal/application/service/tenant_skill_install_test.go index 9b9f73cc3c..e8c03a41d6 100644 --- a/internal/application/service/tenant_skill_install_test.go +++ b/internal/application/service/tenant_skill_install_test.go @@ -310,6 +310,321 @@ func TestInstallSkillRefusesWhenBundleCannotBeStored(t *testing.T) { require.NotContains(t, fx.events, "create-session") } +func TestInstallSkillSkipsWhenReadyWithTheSameArchive(t *testing.T) { + fx := newInstallFixture(t) + archive := zipBundle(t, map[string]string{ + "SKILL.md": validSkillMD, + "scripts/extract.py": "print('hi')\n", + }) + bundle, err := ParseSkillBundle(archive) + require.NoError(t, err) + fx.seedReadySkillWithSHA(bundle.SHA256, "snap-live") + + id, err := fx.svc.InstallSkill(context.Background(), 7, "cfg-1", archive) + + require.NoError(t, err) + require.Equal(t, "sk-1", id) + skill, getErr := fx.skillRepo.GetSkill(context.Background(), 7, "cfg-1", "sk-1") + require.NoError(t, getErr) + require.Equal(t, types.SkillStatusReady, skill.Status, + "a ready skill whose archive did not change must not be flipped to installing") + require.Empty(t, fx.sessionCalls, "the same bytes must not boot a billed sandbox") + require.NotContains(t, fx.events, "create-snapshot") + require.Nil(t, fx.configRepo.saved, "the image pointer must stay where it is") + require.Equal(t, 1, fx.savedBundles, + "a no-op re-upload must still refresh the stored archive for read_skill") + require.Equal(t, "file://bundle.zip", skill.BundleRef) +} + +func TestInstallSkillRetriesAFailedSkillWithTheSameArchive(t *testing.T) { + fx := newInstallFixture(t) + archive := zipBundle(t, map[string]string{ + "SKILL.md": validSkillMD, + "scripts/extract.py": "print('hi')\n", + }) + bundle, err := ParseSkillBundle(archive) + require.NoError(t, err) + require.NoError(t, fx.skillRepo.UpdateSkill(context.Background(), &types.TenantSkillEntity{ + ID: "sk-1", TenantID: 7, SandboxConfigID: "cfg-1", + Name: bundle.Name, BundleSHA256: bundle.SHA256, + Status: types.SkillStatusFailed, Error: "previous run died", + })) + + id, err := fx.svc.InstallSkill(context.Background(), 7, "cfg-1", archive) + + require.NoError(t, err) + require.Equal(t, "sk-1", id) + skill, getErr := fx.skillRepo.GetSkill(context.Background(), 7, "cfg-1", "sk-1") + require.NoError(t, getErr) + require.Equal(t, types.SkillStatusInstalling, skill.Status, + "a failed skill is a retry even when the archive digest is unchanged") +} + +func TestInstallSkillReinstallsWhenTheLiveImageNoLongerCarriesTheSkill(t *testing.T) { + fx := newInstallFixture(t) + archive := zipBundle(t, map[string]string{ + "SKILL.md": validSkillMD, + "scripts/extract.py": "print('hi')\n", + }) + bundle, err := ParseSkillBundle(archive) + require.NoError(t, err) + require.NoError(t, fx.skillRepo.UpdateSkill(context.Background(), &types.TenantSkillEntity{ + ID: "sk-1", TenantID: 7, SandboxConfigID: "cfg-1", + Name: bundle.Name, BundleSHA256: bundle.SHA256, + Status: types.SkillStatusReady, + })) + // The pointer was cleared (last-skill removal, or a rebuild from base). + // The row still says ready, but the files are gone from every new session. + + id, err := fx.svc.InstallSkill(context.Background(), 7, "cfg-1", archive) + + require.NoError(t, err) + require.Equal(t, "sk-1", id) + skill, getErr := fx.skillRepo.GetSkill(context.Background(), 7, "cfg-1", "sk-1") + require.NoError(t, getErr) + require.Equal(t, types.SkillStatusInstalling, skill.Status, + "a ready row whose files left the image is a repair, not a skip") +} + +func TestInstallSkillSkipsAnInFlightInstallOfTheSameArchive(t *testing.T) { + fx := newInstallFixture(t) + archive := zipBundle(t, map[string]string{ + "SKILL.md": validSkillMD, + "scripts/extract.py": "print('hi')\n", + }) + bundle, err := ParseSkillBundle(archive) + require.NoError(t, err) + // One heartbeat ago: the first run is slow, not gone. + beat := fx.now().Add(-skillInstallHeartbeatInterval) + require.NoError(t, fx.skillRepo.UpdateSkill(context.Background(), &types.TenantSkillEntity{ + ID: "sk-1", TenantID: 7, SandboxConfigID: "cfg-1", + Name: bundle.Name, BundleSHA256: bundle.SHA256, + Status: types.SkillStatusInstalling, InstallingSince: &beat, + })) + + id, err := fx.svc.InstallSkill(context.Background(), 7, "cfg-1", archive) + + require.NoError(t, err) + require.Equal(t, "sk-1", id) + require.Empty(t, fx.sessionCalls, + "a second upload of the same bytes must not start another billed run") +} + +// A run that keeps beating is left alone however long it takes: a single agent +// command may take installCommandTimeout, and an install runs several. +func TestInstallSkillSkipsAnInstallThatIsSlowButStillBeating(t *testing.T) { + fx := newInstallFixture(t) + archive := zipBundle(t, map[string]string{ + "SKILL.md": validSkillMD, + "scripts/extract.py": "print('hi')\n", + }) + bundle, err := ParseSkillBundle(archive) + require.NoError(t, err) + submitted := fx.now().Add(-3 * installCommandTimeout) + beat := fx.now().Add(-skillInstallHeartbeatInterval) + require.NoError(t, fx.skillRepo.UpdateSkill(context.Background(), &types.TenantSkillEntity{ + ID: "sk-1", TenantID: 7, SandboxConfigID: "cfg-1", + Name: bundle.Name, BundleSHA256: bundle.SHA256, + Status: types.SkillStatusInstalling, InstallingSince: &beat, + CreatedAt: submitted, + })) + + id, err := fx.svc.InstallSkill(context.Background(), 7, "cfg-1", archive) + + require.NoError(t, err) + require.Equal(t, "sk-1", id) + require.Empty(t, fx.sessionCalls, + "an install that started long ago but is still beating must not be restarted") +} + +func TestInstallSkillRetriesAStaleInFlightInstallOfTheSameArchive(t *testing.T) { + fx := newInstallFixture(t) + archive := zipBundle(t, map[string]string{ + "SKILL.md": validSkillMD, + "scripts/extract.py": "print('hi')\n", + }) + bundle, err := ParseSkillBundle(archive) + require.NoError(t, err) + // The heartbeat stopped: the process that owned this row is gone. + stale := fx.now().Add(-skillInstallInFlightSkip - time.Minute) + require.NoError(t, fx.skillRepo.UpdateSkill(context.Background(), &types.TenantSkillEntity{ + ID: "sk-1", TenantID: 7, SandboxConfigID: "cfg-1", + Name: bundle.Name, BundleSHA256: bundle.SHA256, + Status: types.SkillStatusInstalling, InstallingSince: &stale, + Error: "the previous process is gone", + })) + + id, err := fx.svc.InstallSkill(context.Background(), 7, "cfg-1", archive) + + require.NoError(t, err) + require.Equal(t, "sk-1", id) + skill, getErr := fx.skillRepo.GetSkill(context.Background(), 7, "cfg-1", "sk-1") + require.NoError(t, getErr) + require.Equal(t, types.SkillStatusInstalling, skill.Status) + require.NotNil(t, skill.InstallingSince) + require.Equal(t, fx.now(), *skill.InstallingSince, + "a dead in-flight row must be allowed to start a new run, not wait for the reaper") + require.Empty(t, skill.Error) +} + +// The ledger records which skill an install snapshotted, not which archive, so +// an installing row must never be answered from the image: the files there may +// belong to the previous bundle of the same skill, and skipping would report a +// success that never happened. +func TestCanSkipInstallNeverAnswersAnInstallingRowFromTheImage(t *testing.T) { + fx := newInstallFixture(t) + ctx := context.Background() + fx.seedReadySkillWithSHA(fx.bundle.SHA256, "snap-live") + existing, err := fx.skillRepo.GetSkill(ctx, 7, "cfg-1", "sk-1") + require.NoError(t, err) + stale := fx.now().Add(-skillInstallInFlightSkip - time.Minute) + existing.Status = types.SkillStatusInstalling + existing.InstallingSince = &stale + + require.False(t, fx.svc.canSkipInstall(ctx, existing, fx.bundle), + "a dead install must be retried, not declared done from another bundle's snapshot") +} + +// A ready skill is only skipped when the ledger can actually say the files are +// still in the live image. An unreadable ledger must reinstall rather than +// report a success nobody verified. +func TestCanSkipInstallRequiresAReadableLedger(t *testing.T) { + fx := newInstallFixture(t) + ctx := context.Background() + fx.seedReadySkillWithSHA(fx.bundle.SHA256, "snap-live") + existing, err := fx.skillRepo.GetSkill(ctx, 7, "cfg-1", "sk-1") + require.NoError(t, err) + require.True(t, fx.svc.canSkipInstall(ctx, existing, fx.bundle), + "the same archive of a ready skill still in the image is a no-op") + + fx.skillRepo.listSnapshotsErr = errors.New("ledger unavailable") + + require.False(t, fx.svc.canSkipInstall(ctx, existing, fx.bundle), + "a skip must be earned by a readable ledger, not assumed") +} + +func TestBeatInstallHeartbeatRestampsOnlyAnInstallingRow(t *testing.T) { + fx := newInstallFixture(t) + ctx := context.Background() + stale := fx.now().Add(-time.Hour) + require.NoError(t, fx.svc.updateSkillFields(ctx, 7, "cfg-1", "sk-1", + func(e *types.TenantSkillEntity) { e.InstallingSince = &stale })) + + fx.svc.beatInstallHeartbeat(ctx, 7, "cfg-1", "sk-1") + + skill, err := fx.skillRepo.GetSkill(ctx, 7, "cfg-1", "sk-1") + require.NoError(t, err) + require.Equal(t, fx.now(), *skill.InstallingSince, + "a live install must keep its liveness timestamp current") + + // A finished run's row is no longer this install's to touch: reviving the + // timestamp would hide a ready skill from nothing and a newer upload from + // the reaper. + require.NoError(t, fx.svc.updateSkillFields(ctx, 7, "cfg-1", "sk-1", + func(e *types.TenantSkillEntity) { + e.Status = types.SkillStatusReady + e.InstallingSince = nil + })) + + fx.svc.beatInstallHeartbeat(ctx, 7, "cfg-1", "sk-1") + + skill, err = fx.skillRepo.GetSkill(ctx, 7, "cfg-1", "sk-1") + require.NoError(t, err) + require.Equal(t, types.SkillStatusReady, skill.Status) + require.Nil(t, skill.InstallingSince, + "a row that left the installing status must not be stamped alive again") +} + +func TestStartInstallHeartbeatBeatsUntilStopped(t *testing.T) { + fx := newInstallFixture(t) + ctx := context.Background() + fx.svc.installHeartbeat = time.Millisecond + stale := fx.now().Add(-time.Hour) + require.NoError(t, fx.svc.updateSkillFields(ctx, 7, "cfg-1", "sk-1", + func(e *types.TenantSkillEntity) { e.InstallingSince = &stale })) + + stop := fx.svc.startInstallHeartbeat(ctx, 7, "cfg-1", "sk-1") + require.Eventually(t, func() bool { + skill, err := fx.skillRepo.GetSkill(ctx, 7, "cfg-1", "sk-1") + return err == nil && skill.InstallingSince != nil && skill.InstallingSince.Equal(fx.now()) + }, 2*time.Second, time.Millisecond, "the heartbeat must restamp the row while the run works") + stop() + + // Stopping is what lets the terminal write stand: a beat landing after it + // would put a serving skill back to installing. + require.NoError(t, fx.svc.updateSkillFields(ctx, 7, "cfg-1", "sk-1", + func(e *types.TenantSkillEntity) { + e.Status = types.SkillStatusReady + e.InstallingSince = nil + })) + time.Sleep(20 * time.Millisecond) + skill, err := fx.skillRepo.GetSkill(ctx, 7, "cfg-1", "sk-1") + require.NoError(t, err) + require.Equal(t, types.SkillStatusReady, skill.Status) + require.Nil(t, skill.InstallingSince) + stop() +} + +func TestInstallSkillDoesNotSkipARemovalOfTheSameArchive(t *testing.T) { + fx := newInstallFixture(t) + archive := zipBundle(t, map[string]string{ + "SKILL.md": validSkillMD, + "scripts/extract.py": "print('hi')\n", + }) + bundle, err := ParseSkillBundle(archive) + require.NoError(t, err) + require.NoError(t, fx.skillRepo.UpdateSkill(context.Background(), &types.TenantSkillEntity{ + ID: "sk-1", TenantID: 7, SandboxConfigID: "cfg-1", + Name: bundle.Name, BundleSHA256: bundle.SHA256, + Status: types.SkillStatusRemoving, + })) + + id, err := fx.svc.InstallSkill(context.Background(), 7, "cfg-1", archive) + + require.NoError(t, err) + require.Equal(t, "sk-1", id) + skill, getErr := fx.skillRepo.GetSkill(context.Background(), 7, "cfg-1", "sk-1") + require.NoError(t, getErr) + require.Equal(t, types.SkillStatusInstalling, skill.Status, + "re-uploading during a removal is how the upload cancels it") +} + +func TestInstallSkillSkipRefusesToPretendSuccessWhenBundleCannotBeStored(t *testing.T) { + fx := newInstallFixture(t) + archive := zipBundle(t, map[string]string{ + "SKILL.md": validSkillMD, + "scripts/extract.py": "print('hi')\n", + }) + bundle, err := ParseSkillBundle(archive) + require.NoError(t, err) + fx.seedReadySkillWithSHA(bundle.SHA256, "snap-live") + fx.saveErr = errors.New("object store down") + + _, err = fx.svc.InstallSkill(context.Background(), 7, "cfg-1", archive) + + require.Error(t, err) + require.ErrorContains(t, err, "store bundle") + skill, getErr := fx.skillRepo.GetSkill(context.Background(), 7, "cfg-1", "sk-1") + require.NoError(t, getErr) + require.Equal(t, types.SkillStatusReady, skill.Status, + "a storage failure on a no-op re-upload must not flip a serving skill to failed") + require.Empty(t, fx.sessionCalls) +} + +func TestRunInstallAbortsWhenTheSameArchiveIsAlreadyServing(t *testing.T) { + fx := newInstallFixture(t) + fx.seedReadySkillWithSHA(fx.bundle.SHA256, "snap-live") + + require.NoError(t, fx.svc.runInstall(context.Background(), 7, "cfg-1", "sk-1", fx.bundle)) + + require.NotContains(t, fx.events, "create-snapshot", + "a sibling retry that lost the race to the first run must not grow another snapshot") + require.Nil(t, fx.configRepo.saved) + skill, err := fx.skillRepo.GetSkill(context.Background(), 7, "cfg-1", "sk-1") + require.NoError(t, err) + require.Equal(t, types.SkillStatusReady, skill.Status) +} + func TestTenantForStoragePrefersMatchingContextTenant(t *testing.T) { svc := &TenantSkillService{} backendID := "backend-1" @@ -833,7 +1148,8 @@ type installFixture struct { engineModel chat.Chat // saveErr fails bundle storage so InstallSkill cannot accept a skill // whose archive will later be unreadable. - saveErr error + saveErr error + savedBundles int } func newInstallFixture(t *testing.T) *installFixture { @@ -905,6 +1221,10 @@ func (f *installFixture) record(event string) { f.events = append(f.events, event) } +// now is the fixture's clock, so a test can express "one heartbeat ago" +// against the same instant the service reads. +func (f *installFixture) now() time.Time { return f.svc.now() } + // seedInstalledSkill puts the fixture in the state a removal starts from: the // skill is ready inside the config's current image, the ledger holds the active // row that produced that image, and the image manifest lists the skill. @@ -945,6 +1265,33 @@ func (f *installFixture) seedInstalledSkill(skillID, snapshotID string, generati f.sandboxMgr.manifest = payload } +// seedReadySkillWithSHA puts the fixture in the state a no-op re-upload starts +// from: the skill is ready, the digest matches the archive about to be posted, +// and the ledger says those files are still on the live image. +func (f *installFixture) seedReadySkillWithSHA(sha256, snapshotID string) { + f.t.Helper() + ctx := context.Background() + skill, err := f.skillRepo.GetSkill(ctx, 7, "cfg-1", "sk-1") + require.NoError(f.t, err) + require.NotNil(f.t, skill) + skill.Status = types.SkillStatusReady + skill.BundleSHA256 = sha256 + skill.InstalledSnapshotID = snapshotID + skill.Error = "" + skill.InstallingSince = nil + require.NoError(f.t, f.skillRepo.UpdateSkill(ctx, skill)) + + require.NoError(f.t, f.skillRepo.CreateSnapshotRow(ctx, &types.TenantSkillSnapshotEntity{ + ID: "row-live", TenantID: 7, SandboxConfigID: "cfg-1", SkillID: "sk-1", + SnapshotID: snapshotID, Generation: 1, + Trigger: types.SkillSnapshotTriggerInstall, State: types.SkillSnapshotStateActive, + })) + f.configRepo.entity.Config.SkillImage = &types.SkillImageConfig{ + SnapshotID: snapshotID, Generation: 1, + BaseTemplateID: "base-template", OwnerFingerprint: f.fingerprint, + } +} + type installConfigRepo struct { fx *installFixture entity *types.TenantSandboxConfigEntity @@ -1046,6 +1393,9 @@ type installSkillRepo struct { // snapshot failing for one state, leaving the snapshot ID nowhere but a // local variable. markStateFails func(state string) bool + // listSnapshotsErr models an unreadable ledger, which is what stands + // between "the image still carries this skill" and a guess. + listSnapshotsErr error // deleteSkillErr models the row delete failing past the point of no // return. deleteSkillErr error @@ -1215,6 +1565,9 @@ func (r *installSkillRepo) ListSnapshotsByConfig( ) ([]*types.TenantSkillSnapshotEntity, error) { r.mu.Lock() defer r.mu.Unlock() + if r.listSnapshotsErr != nil { + return nil, r.listSnapshotsErr + } var out []*types.TenantSkillSnapshotEntity for _, e := range r.snapshots { if e.TenantID == tenantID && e.SandboxConfigID == configID { @@ -1751,8 +2104,11 @@ func (installFileService) SaveFile(context.Context, *multipart.FileHeader, uint6 } func (s installFileService) SaveBytes(context.Context, []byte, uint64, string, bool) (string, error) { - if s.fx != nil && s.fx.saveErr != nil { - return "", s.fx.saveErr + if s.fx != nil { + s.fx.savedBundles++ + if s.fx.saveErr != nil { + return "", s.fx.saveErr + } } return "file://bundle.zip", nil } diff --git a/internal/application/service/tenant_skill_reaper.go b/internal/application/service/tenant_skill_reaper.go index ed16f4e974..882351bb68 100644 --- a/internal/application/service/tenant_skill_reaper.go +++ b/internal/application/service/tenant_skill_reaper.go @@ -47,6 +47,13 @@ type skillSnapshotLister interface { ListSnapshots(ctx context.Context, sandboxID string) ([]sandbox.RemoteSnapshotRef, error) } +// skillSnapshotDeleter is the provider delete PruneSupersededSnapshots is +// allowed to call. It is a separate surface from the lister so reconcile +// cannot grow a delete by accident: extras not in the ledger stay untouched. +type skillSnapshotDeleter interface { + DeleteSnapshot(ctx context.Context, snapshotID string) error +} + // sandboxConfigEnumerator walks every sandbox config for the orphan-snapshot // sweep. ListAll is housekeeping-only. type sandboxConfigEnumerator interface { @@ -59,6 +66,7 @@ var ( _ skillReaperConfigReader = (repository.TenantSandboxConfigRepository)(nil) _ sandboxConfigEnumerator = (repository.TenantSandboxConfigRepository)(nil) _ skillSnapshotLister = (sandbox.RemoteSnapshotManager)(nil) + _ skillSnapshotDeleter = (*sandbox.SessionBoundManager)(nil) ) // ReapStuckRuns recovers skill rows whose install or remove process died. @@ -67,11 +75,12 @@ var ( // still carry this skill's files — and skillFilesInLiveImage answers it from // the snapshot ledger rather than from the row. // -// An installing row older than skillInstallStuckTTL is healed to ready when -// the files are there: a re-install that died before the pointer moved, or a -// terminal ready write that never landed. Leaving it at installing would hide -// a skill the image still carries. Otherwise it becomes failed so the UI stops -// spinning. +// An installing row whose heartbeat has been silent for skillInstallStuckTTL +// is healed to ready when the files are there: a re-install that died before +// the pointer moved, or a terminal ready write that never landed. Leaving it +// at installing would hide a skill the image still carries. Otherwise it +// becomes failed so the UI stops spinning. A live install keeps stamping +// InstallingSince, so a run that is merely slow is never swept. // // A removing row is restored to ready while the files are still there, so the // operator can retry. Once they are gone the leftover row is deleted, so the @@ -80,11 +89,7 @@ func (s *TenantSkillService) ReapStuckRuns(ctx context.Context) (int, error) { if s == nil || s.skills == nil { return 0, nil } - now := s.now - if now == nil { - now = time.Now - } - cutoff := now().Add(-skillInstallStuckTTL) + cutoff := s.clock()().Add(-skillInstallStuckTTL) stale, err := s.skills.ListStaleInstalling(ctx, cutoff) if err != nil { return 0, err @@ -344,6 +349,204 @@ func snapshotListerFrom( return lister } +func snapshotDeleterFrom( + ctx context.Context, resolver sandbox.TenantSandboxResolver, tenantID uint64, configID string, +) skillSnapshotDeleter { + if resolver == nil { + return nil + } + mgr, err := resolver.Resolve(ctx, tenantID, configID) + if err != nil { + logger.Warnf(ctx, "[skill] resolve sandbox for snapshot prune of %s failed: %v", configID, err) + return nil + } + if mgr == nil { + return nil + } + deleter, ok := mgr.(skillSnapshotDeleter) + if !ok { + return nil + } + return deleter +} + +func (s *TenantSkillService) snapshotRetentionWindow() time.Duration { + if s != nil && s.snapshotRetention > 0 { + return s.snapshotRetention + } + return skillSnapshotRetention +} + +// snapshotRetentionFor is how long this config's retired snapshots stay on +// the provider. The floor is snapshotRetentionWindow; a config that asked +// for a sandbox TTL longer than that keeps the previous template at least +// that long plus a margin, so a session created from it can still exist. +func (s *TenantSkillService) snapshotRetentionFor(cfg *types.TenantSandboxConfigEntity) time.Duration { + window := s.snapshotRetentionWindow() + ttl := time.Duration(0) + if cfg != nil && cfg.Config != nil { + ttl = configuredSandboxTTL(cfg.Config) + } + if needed := ttl + skillSnapshotTTLMargin; needed > window { + return needed + } + return window +} + +func configuredSandboxTTL(cfg *types.TenantSandboxConfig) time.Duration { + if cfg == nil { + return 0 + } + seconds := 0 + if cfg.Cube != nil && cfg.Cube.CubeSandboxTTLSeconds > seconds { + seconds = cfg.Cube.CubeSandboxTTLSeconds + } + if cfg.E2B != nil && cfg.E2B.E2BSandboxTTLSeconds > seconds { + seconds = cfg.E2B.E2BSandboxTTLSeconds + } + if seconds <= 0 { + return 0 + } + return time.Duration(seconds) * time.Second +} + +// PruneSupersededSnapshots deletes provider snapshots the ledger has already +// retired, once they are older than this config's retention. The current +// image is never touched, nor is anything the ledger does not name: extras +// belong to other environments on a shared provider account. +// +// A live sandbox does not need the template it was created from in order to +// keep running, so the only reason to wait is in-flight creates that resolved +// the previous pointer. Twenty-four hours is far past every backend's default +// TTL; a config that sets a longer one extends the wait. +func (s *TenantSkillService) PruneSupersededSnapshots(ctx context.Context) (int, error) { + if s == nil || s.skills == nil { + return 0, nil + } + enum, ok := s.configs.(sandboxConfigEnumerator) + if !ok { + return 0, nil + } + configs, err := enum.ListAll(ctx) + if err != nil { + return 0, err + } + now := s.clock() + pruned := 0 + for _, cfg := range configs { + if cfg == nil || types.IsSandboxWorkspacePolicyRow(cfg) { + continue + } + cutoff := now().Add(-s.snapshotRetentionFor(cfg)) + n, err := s.pruneConfigSnapshots(ctx, cfg, cutoff) + if err != nil { + logger.Warnf(ctx, "[skill] prune superseded snapshots of config %s failed: %v", + cfg.ID, err) + continue + } + pruned += n + } + return pruned, nil +} + +func (s *TenantSkillService) pruneConfigSnapshots( + ctx context.Context, cfg *types.TenantSandboxConfigEntity, cutoff time.Time, +) (int, error) { + if cfg == nil { + return 0, nil + } + // A rotated credential points at a different provider account, where the + // ledger's snapshot IDs do not exist. The delete would come back + // not-found, which this sweep reads as "already gone", so the account that + // really holds those snapshots would keep being billed for them while the + // ledger recorded them as deleted. ensureUsableImage stops installs for + // the same reason; this stops the irreversible half. + if err := ensureUsableImage(cfg); err != nil { + logger.Warnf(ctx, "[skill] skip snapshot prune of config %s: %v", cfg.ID, err) + return 0, nil + } + rows, err := s.skills.ListSnapshotsByConfig(ctx, cfg.TenantID, cfg.ID) + if err != nil { + return 0, err + } + live := currentSnapshotID(cfg) + // Resolving builds a provider client, so it waits until a row is actually + // eligible: most configs have nothing to prune on most sweeps. + var deleter skillSnapshotDeleter + resolved := false + pruned := 0 + for _, row := range rows { + if !snapshotEligibleForPrune(row, live, cutoff) { + continue + } + if !resolved { + deleter = snapshotDeleterFrom(ctx, s.sandboxes, cfg.TenantID, cfg.ID) + resolved = true + } + if deleter == nil { + logger.Warnf(ctx, + "[skill] cannot prune snapshot %s of config %s: provider does not support delete", + row.SnapshotID, cfg.ID) + continue + } + if err := deleter.DeleteSnapshot(ctx, row.SnapshotID); err != nil && !sandbox.IsRemoteNotFound(err) { + logger.Warnf(ctx, "[skill] delete superseded snapshot %s failed: %v", row.SnapshotID, err) + continue + } + if err := s.skills.MarkSnapshotState( + ctx, cfg.TenantID, row.ID, types.SkillSnapshotStateDeleted, row.SnapshotID, + ); err != nil { + logger.Warnf(ctx, "[skill] mark snapshot %s deleted after prune failed: %v", row.ID, err) + continue + } + pruned++ + } + return pruned, nil +} + +// snapshotEligibleForPrune is the ledger-side gate. The provider delete is +// what costs money; this is what keeps it from touching the live image or a +// snapshot another environment created on the same account. +// +// Superseded rows are the normal case. Active rows that are not the live +// pointer are the crash window between switchImagePointer and +// markPreviousSnapshotsSuperseded: they are billed leftovers too, aged from +// UpdatedAt / CreatedAt because they never got a SupersededAt. +func snapshotEligibleForPrune( + row *types.TenantSkillSnapshotEntity, liveSnapshotID string, cutoff time.Time, +) bool { + if row == nil { + return false + } + id := strings.TrimSpace(row.SnapshotID) + if id == "" || id == strings.TrimSpace(liveSnapshotID) { + return false + } + switch row.State { + case types.SkillSnapshotStateSuperseded, types.SkillSnapshotStateActive: + default: + return false + } + aged := snapshotPruneAge(row) + return aged != nil && aged.Before(cutoff) +} + +func snapshotPruneAge(row *types.TenantSkillSnapshotEntity) *time.Time { + if row == nil { + return nil + } + if row.State == types.SkillSnapshotStateSuperseded && row.SupersededAt != nil { + return row.SupersededAt + } + if !row.UpdatedAt.IsZero() { + return &row.UpdatedAt + } + if !row.CreatedAt.IsZero() { + return &row.CreatedAt + } + return nil +} + func (s *TenantSkillService) reconcileAllSnapshots(ctx context.Context) { enum, ok := s.configs.(sandboxConfigEnumerator) if !ok { @@ -368,6 +571,9 @@ func (s *TenantSkillService) runSkillReaper(ctx context.Context) { if _, err := s.ReapStuckRuns(ctx); err != nil { logger.Warnf(ctx, "[skill] reap stuck runs failed: %v", err) } + if _, err := s.PruneSupersededSnapshots(ctx); err != nil { + logger.Warnf(ctx, "[skill] prune superseded snapshots failed: %v", err) + } s.reconcileAllSnapshots(ctx) } diff --git a/internal/application/service/tenant_skill_reaper_test.go b/internal/application/service/tenant_skill_reaper_test.go index 4b2aa544fc..49cb6c4262 100644 --- a/internal/application/service/tenant_skill_reaper_test.go +++ b/internal/application/service/tenant_skill_reaper_test.go @@ -2,6 +2,7 @@ package service import ( "context" + "errors" "strings" "testing" "time" @@ -234,6 +235,200 @@ func TestReconcileSnapshotsWarnsExtrasWithoutDeleting(t *testing.T) { "extras are only warned; the same provider account may be shared across environments") } +func TestPruneSupersededSnapshotsDeletesOldLedgerSnapshots(t *testing.T) { + fx := newReaperFixture(t) + fx.svc.snapshotRetention = time.Hour + old := fx.now.Add(-2 * time.Hour) + recent := fx.now.Add(-30 * time.Minute) + fx.live("snap-live") + fx.superseded("sk-1", "snap-old", "", old) + fx.superseded("sk-2", "snap-recent", "snap-old", recent) + fx.installed("sk-3", "snap-live", "snap-recent") + fx.provider.listed = []sandbox.RemoteSnapshotRef{ + {ID: "snap-old"}, {ID: "snap-recent"}, {ID: "snap-live"}, {ID: "snap-foreign"}, + } + + n, err := fx.svc.PruneSupersededSnapshots(context.Background()) + + require.NoError(t, err) + require.Equal(t, 1, n) + require.Equal(t, []string{"snap-old"}, fx.provider.deleted, + "only a superseded snapshot older than retention is a billed leftover") + rows, err := fx.skills.ListSnapshotsByConfig(context.Background(), 7, "cfg-1") + require.NoError(t, err) + states := map[string]string{} + for _, row := range rows { + states[row.SnapshotID] = row.State + } + require.Equal(t, types.SkillSnapshotStateDeleted, states["snap-old"]) + require.Equal(t, types.SkillSnapshotStateSuperseded, states["snap-recent"], + "a snapshot still inside the retention window may have live sandboxes") + require.Equal(t, types.SkillSnapshotStateActive, states["snap-live"]) +} + +func TestPruneSupersededSnapshotsNeverDeletesTheLiveImage(t *testing.T) { + fx := newReaperFixture(t) + fx.svc.snapshotRetention = time.Hour + old := fx.now.Add(-2 * time.Hour) + fx.live("snap-live") + fx.skills.snapshots = append(fx.skills.snapshots, &types.TenantSkillSnapshotEntity{ + ID: "row-wrong", TenantID: 7, SandboxConfigID: "cfg-1", SkillID: "sk-1", + SnapshotID: "snap-live", Trigger: types.SkillSnapshotTriggerInstall, + State: types.SkillSnapshotStateSuperseded, SupersededAt: &old, + }) + + n, err := fx.svc.PruneSupersededSnapshots(context.Background()) + + require.NoError(t, err) + require.Zero(t, n) + require.Empty(t, fx.provider.deleted, + "a ledger bug that marks the live snapshot superseded must not delete it") +} + +func TestPruneSupersededSnapshotsNeverDeletesUnknownProviderSnapshots(t *testing.T) { + fx := newReaperFixture(t) + fx.svc.snapshotRetention = time.Hour + fx.live("snap-live") + fx.installed("sk-1", "snap-live", "") + fx.provider.listed = []sandbox.RemoteSnapshotRef{ + {ID: "snap-live"}, {ID: "snap-foreign"}, + } + + n, err := fx.svc.PruneSupersededSnapshots(context.Background()) + + require.NoError(t, err) + require.Zero(t, n) + require.Empty(t, fx.provider.deleted, + "a snapshot the ledger does not name belongs to another environment") +} + +func TestPruneSupersededSnapshotsDeletesStaleActiveLeftovers(t *testing.T) { + fx := newReaperFixture(t) + fx.svc.snapshotRetention = time.Hour + old := fx.now.Add(-2 * time.Hour) + fx.live("snap-live") + fx.installed("sk-2", "snap-live", "snap-old") + fx.skills.snapshots = append(fx.skills.snapshots, &types.TenantSkillSnapshotEntity{ + ID: "row-snap-old", TenantID: 7, SandboxConfigID: "cfg-1", SkillID: "sk-1", + SnapshotID: "snap-old", Trigger: types.SkillSnapshotTriggerInstall, + State: types.SkillSnapshotStateActive, CreatedAt: old, UpdatedAt: old, + }) + + n, err := fx.svc.PruneSupersededSnapshots(context.Background()) + + require.NoError(t, err) + require.Equal(t, 1, n) + require.Equal(t, []string{"snap-old"}, fx.provider.deleted, + "a pointer switch that never marked the previous row superseded still leaves a billed snapshot") + rows, err := fx.skills.ListSnapshotsByConfig(context.Background(), 7, "cfg-1") + require.NoError(t, err) + states := map[string]string{} + for _, row := range rows { + states[row.SnapshotID] = row.State + } + require.Equal(t, types.SkillSnapshotStateDeleted, states["snap-old"]) + require.Equal(t, types.SkillSnapshotStateActive, states["snap-live"]) +} + +// A rotated credential points at another provider account, where these IDs do +// not exist. The delete would come back not-found, which the sweep reads as +// "already gone", so the account that really holds them would keep being +// billed while the ledger recorded them as deleted. +func TestPruneSupersededSnapshotsSkipsAConfigBuiltByAnotherAccount(t *testing.T) { + fx := newReaperFixture(t) + fx.svc.snapshotRetention = time.Hour + old := fx.now.Add(-2 * time.Hour) + fx.live("snap-live") + fx.superseded("sk-1", "snap-old", "", old) + fx.configs.entity.Config.E2B.APIKey = "rotated-key" + + n, err := fx.svc.PruneSupersededSnapshots(context.Background()) + + require.NoError(t, err) + require.Zero(t, n) + require.Empty(t, fx.provider.deleted, + "snapshots of an account we can no longer address must not be recorded as deleted") + rows, err := fx.skills.ListSnapshotsByConfig(context.Background(), 7, "cfg-1") + require.NoError(t, err) + require.Equal(t, types.SkillSnapshotStateSuperseded, rows[0].State) +} + +// Resolving a provider builds a client. Most configs have nothing to prune on +// most sweeps, and the sweep runs every five minutes across every workspace. +func TestPruneSupersededSnapshotsBuildsNoProviderClientWithNothingToPrune(t *testing.T) { + fx := newReaperFixture(t) + fx.svc.snapshotRetention = time.Hour + recent := fx.now.Add(-30 * time.Minute) + fx.live("snap-live") + fx.installed("sk-1", "snap-live", "snap-recent") + fx.superseded("sk-2", "snap-recent", "", recent) + + n, err := fx.svc.PruneSupersededSnapshots(context.Background()) + + require.NoError(t, err) + require.Zero(t, n) + require.Zero(t, fx.resolver.resolves, + "a sweep with no eligible row must not pay for a provider client") +} + +func TestPruneSupersededSnapshotsLeavesTheRowWhenDeleteFails(t *testing.T) { + fx := newReaperFixture(t) + fx.svc.snapshotRetention = time.Hour + old := fx.now.Add(-2 * time.Hour) + fx.live("snap-live") + fx.superseded("sk-1", "snap-old", "", old) + fx.provider.deleteErr = errors.New("provider down") + + n, err := fx.svc.PruneSupersededSnapshots(context.Background()) + + require.NoError(t, err) + require.Zero(t, n) + require.Empty(t, fx.provider.deleted) + rows, err := fx.skills.ListSnapshotsByConfig(context.Background(), 7, "cfg-1") + require.NoError(t, err) + require.Equal(t, types.SkillSnapshotStateSuperseded, rows[0].State, + "a failed provider delete must not be recorded as deleted") +} + +func TestPruneSupersededSnapshotsTreatsMissingProviderSnapshotAsDeleted(t *testing.T) { + fx := newReaperFixture(t) + fx.svc.snapshotRetention = time.Hour + old := fx.now.Add(-2 * time.Hour) + fx.live("snap-live") + fx.superseded("sk-1", "snap-old", "", old) + fx.provider.deleteErr = &sandbox.RemoteError{ + Kind: sandbox.RemoteErrorKindNotFound, Op: "DeleteSnapshot", + } + + n, err := fx.svc.PruneSupersededSnapshots(context.Background()) + + require.NoError(t, err) + require.Equal(t, 1, n) + rows, err := fx.skills.ListSnapshotsByConfig(context.Background(), 7, "cfg-1") + require.NoError(t, err) + require.Equal(t, types.SkillSnapshotStateDeleted, rows[0].State, + "a snapshot the provider already dropped is gone; the ledger must catch up") +} + +func TestPruneSupersededSnapshotsHonoursALongerConfiguredSandboxTTL(t *testing.T) { + fx := newReaperFixture(t) + fx.svc.snapshotRetention = time.Hour + fx.configs.entity.Config.E2B.E2BSandboxTTLSeconds = int((48 * time.Hour).Seconds()) + young := fx.now.Add(-25 * time.Hour) + old := fx.now.Add(-50 * time.Hour) + fx.live("snap-live") + fx.superseded("sk-1", "snap-young", "", young) + fx.superseded("sk-2", "snap-old", "snap-young", old) + fx.installed("sk-3", "snap-live", "snap-old") + + n, err := fx.svc.PruneSupersededSnapshots(context.Background()) + + require.NoError(t, err) + require.Equal(t, 1, n) + require.Equal(t, []string{"snap-old"}, fx.provider.deleted, + "a config whose sandbox TTL exceeds the floor must keep templates that young") +} + func TestTenantSkillServiceStartIsIdempotent(t *testing.T) { svc := NewTenantSkillService(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil) @@ -249,28 +444,45 @@ type reaperFixture struct { skills *reaperSkillStore configs *reaperConfigStore provider *reaperSnapshotProvider - now time.Time + resolver *reaperSandboxResolver + // fingerprint is the credential identity the stored image was built with. + // A prune that could not tell it from another account's would delete + // snapshots it cannot even see, so the fixture carries a real one. + fingerprint string + now time.Time } func newReaperFixture(t *testing.T) *reaperFixture { t.Helper() now := time.Date(2026, 8, 20, 12, 0, 0, 0, time.UTC) + fingerprint := sandbox.SkillImageFingerprint("e2b", "key-1", "https://e2b.example") skills := &reaperSkillStore{rows: map[string]*types.TenantSkillEntity{}} configs := &reaperConfigStore{ entity: &types.TenantSandboxConfigEntity{ ID: "cfg-1", TenantID: 7, + SandboxType: string(sandbox.SandboxTypeE2B), Config: &types.TenantSandboxConfig{ - SkillImage: &types.SkillImageConfig{SnapshotID: "snap-other"}, + SandboxType: string(sandbox.SandboxTypeE2B), + E2B: &types.E2BSandboxConfig{ + APIURL: "https://e2b.example", APIKey: "key-1", TemplateID: "base-template", + }, + SkillImage: &types.SkillImageConfig{ + SnapshotID: "snap-other", OwnerFingerprint: fingerprint, + }, }, }, } provider := &reaperSnapshotProvider{} + resolver := &reaperSandboxResolver{provider: provider} svc := NewTenantSkillService( - skills, configs, nil, &reaperSandboxResolver{provider: provider}, + skills, configs, nil, resolver, nil, nil, nil, nil, nil, nil, nil, nil, ) svc.now = func() time.Time { return now } - return &reaperFixture{svc: svc, skills: skills, configs: configs, provider: provider, now: now} + return &reaperFixture{ + svc: svc, skills: skills, configs: configs, provider: provider, + resolver: resolver, fingerprint: fingerprint, now: now, + } } // installed and removed write the ledger row an install or a removal leaves @@ -291,10 +503,21 @@ func (f *reaperFixture) snapshotRow(skillID, snapshotID, parentSnapshotID, trigg }) } +func (f *reaperFixture) superseded(skillID, snapshotID, parentSnapshotID string, at time.Time) { + f.skills.snapshots = append(f.skills.snapshots, &types.TenantSkillSnapshotEntity{ + ID: "row-" + snapshotID, TenantID: 7, SandboxConfigID: "cfg-1", SkillID: skillID, + SnapshotID: snapshotID, ParentSnapshotID: parentSnapshotID, + Trigger: types.SkillSnapshotTriggerInstall, State: types.SkillSnapshotStateSuperseded, + SupersededAt: &at, + }) +} + // live points the config at a snapshot, the way an install's pointer switch // does. func (f *reaperFixture) live(snapshotID string) { - f.configs.entity.Config.SkillImage = &types.SkillImageConfig{SnapshotID: snapshotID} + f.configs.entity.Config.SkillImage = &types.SkillImageConfig{ + SnapshotID: snapshotID, OwnerFingerprint: f.fingerprint, + } } var ( @@ -304,6 +527,7 @@ var ( _ sandboxConfigEnumerator = (*reaperConfigStore)(nil) _ sandbox.TenantSandboxResolver = (*reaperSandboxResolver)(nil) _ skillSnapshotLister = (*reaperSnapshotProvider)(nil) + _ skillSnapshotDeleter = (*reaperSnapshotProvider)(nil) ) type reaperSkillStore struct { @@ -395,9 +619,23 @@ func (r *reaperSkillStore) CreateSnapshotRow(_ context.Context, e *types.TenantS } func (r *reaperSkillStore) MarkSnapshotState( - context.Context, uint64, string, string, string, + _ context.Context, tenantID uint64, id, state, snapshotID string, ) error { - panic("MarkSnapshotState is outside the reaper surface") + for _, e := range r.snapshots { + if e == nil || e.ID != id || e.TenantID != tenantID { + continue + } + e.State = state + if snapshotID != "" { + e.SnapshotID = snapshotID + } + if state == types.SkillSnapshotStateSuperseded { + now := time.Now() + e.SupersededAt = &now + } + return nil + } + return nil } func (r *reaperSkillStore) DeleteSnapshotRowsByConfig(context.Context, uint64, string) error { @@ -452,9 +690,13 @@ func (r *reaperConfigStore) ClearCordon(context.Context, uint64, string) error { type reaperSandboxResolver struct { provider *reaperSnapshotProvider + // resolves counts provider constructions. Resolving builds a client, so a + // sweep with nothing to do must not pay for one. + resolves int } func (r *reaperSandboxResolver) Resolve(context.Context, uint64, string) (sandbox.Manager, error) { + r.resolves++ return r.provider, nil } @@ -464,6 +706,7 @@ type reaperSnapshotProvider struct { listed []sandbox.RemoteSnapshotRef listCalls []string deleted []string + deleteErr error } func (p *reaperSnapshotProvider) ListSnapshots( @@ -474,6 +717,9 @@ func (p *reaperSnapshotProvider) ListSnapshots( } func (p *reaperSnapshotProvider) DeleteSnapshot(_ context.Context, snapshotID string) error { + if p.deleteErr != nil { + return p.deleteErr + } p.deleted = append(p.deleted, snapshotID) return nil } diff --git a/internal/application/service/tenant_skill_service.go b/internal/application/service/tenant_skill_service.go index ed4c0d3fa5..8da4e70d52 100644 --- a/internal/application/service/tenant_skill_service.go +++ b/internal/application/service/tenant_skill_service.go @@ -19,9 +19,39 @@ import ( // lock without renewing. Installs run for minutes, so the lease is renewed by // redislock rather than being set long. const ( - skillImageLockLease = 30 * time.Second - skillImageLockRenew = 10 * time.Second + skillImageLockLease = 30 * time.Second + skillImageLockRenew = 10 * time.Second + + // skillInstallStuckTTL is how long a run may go without a heartbeat + // before the reaper treats it as abandoned. It is a silence budget, not + // a duration budget: a legitimate install that spends two hours in the + // agent keeps beating and is left alone. skillInstallStuckTTL = 60 * time.Minute + + // skillInstallHeartbeatInterval is how often a running install stamps + // InstallingSince to say its process is still alive. Everything that has + // to tell "still working" from "died" reads that timestamp. + skillInstallHeartbeatInterval = 30 * time.Second + + // skillInstallInFlightSkip is how much heartbeat silence makes a second + // upload of the same archive stop deferring to the run that owns the row. + // It is a multiple of the heartbeat so a slow install is never mistaken + // for a dead one, and short enough that a re-upload recovers a dead + // process in minutes instead of waiting for skillInstallStuckTTL. + skillInstallInFlightSkip = 3 * time.Minute + + // skillSnapshotRetention is how long a superseded snapshot stays on the + // provider after the pointer has moved. Live sandboxes may still have + // been created from it (especially SkillRolloutNewSession); once they + // expire, the template is only a billed leftover. Twenty-four hours is + // well past every backend's default sandbox TTL. A config that sets a + // longer sandbox TTL extends this via snapshotRetentionFor. + skillSnapshotRetention = 24 * time.Hour + + // skillSnapshotTTLMargin is added on top of a config's own sandbox TTL + // so an in-flight create that resolved the previous pointer still has + // a template to boot from. + skillSnapshotTTLMargin = time.Hour ) // TenantSkillService owns the skill image lifecycle for sandbox configs. @@ -52,6 +82,14 @@ type TenantSkillService struct { // test can let an install outlast it, which every real install does. cleanupTimeout time.Duration + // snapshotRetention is how long a superseded snapshot is kept on the + // provider. Injectable so a prune test can age a row without waiting a day. + snapshotRetention time.Duration + + // installHeartbeat is how often a running install restamps its liveness. + // Injectable so a test can observe a beat without waiting half a minute. + installHeartbeat time.Duration + // localLocks serialises installs when Redis is absent. It only guards this // process; multi-replica deployments require Redis for cross-process safety. localLocks *keyedMutex @@ -79,21 +117,23 @@ func NewTenantSkillService( messages interfaces.MessageRepository, ) *TenantSkillService { return &TenantSkillService{ - skills: skillsRepo, - configs: configsRepo, - resolver: resolver, - sandboxes: sandboxes, - sandboxPolicy: sandboxPolicy, - agents: agents, - installerAgents: customAgents, - sessions: sessions, - models: models, - redis: redisClient, - streams: streams, - messages: messages, - now: time.Now, - cleanupTimeout: installCleanupTimeout, - localLocks: newKeyedMutex(), + skills: skillsRepo, + configs: configsRepo, + resolver: resolver, + sandboxes: sandboxes, + sandboxPolicy: sandboxPolicy, + agents: agents, + installerAgents: customAgents, + sessions: sessions, + models: models, + redis: redisClient, + streams: streams, + messages: messages, + now: time.Now, + cleanupTimeout: installCleanupTimeout, + snapshotRetention: skillSnapshotRetention, + installHeartbeat: skillInstallHeartbeatInterval, + localLocks: newKeyedMutex(), cron: cron.New(cron.WithSeconds(), cron.WithChain( cron.Recover(cron.DefaultLogger), )), @@ -127,6 +167,15 @@ func skillImageLockKey(tenantID uint64, configID string) string { return fmt.Sprintf("weknora-skill-image-lock:%d:%s", tenantID, configID) } +// clock is this service's time source. Tests inject one; a service built +// without NewTenantSkillService still gets a working default. +func (s *TenantSkillService) clock() func() time.Time { + if s != nil && s.now != nil { + return s.now + } + return time.Now +} + // keyedMutex is the no-Redis fallback for withConfigLock. type keyedMutex struct { mu sync.Mutex diff --git a/internal/sandbox/session_manager.go b/internal/sandbox/session_manager.go index 0be39fd9a1..3b503fae97 100644 --- a/internal/sandbox/session_manager.go +++ b/internal/sandbox/session_manager.go @@ -426,8 +426,9 @@ func (m *SessionBoundManager) CreateSnapshot( return snapshots.CreateSnapshot(ctx, handle.ID(), name) } -// DeleteSnapshot forwards provider snapshot deletion. It is used only to clean -// up a just-created orphan when the DB pointer switch fails. +// DeleteSnapshot forwards provider snapshot deletion. The skill install path +// uses it to abandon an orphan when the pointer switch fails; the reaper uses +// it to prune superseded snapshots that have aged past retention. func (m *SessionBoundManager) DeleteSnapshot(ctx context.Context, snapshotID string) error { if err := m.requireRemoteBackend(); err != nil { return err diff --git a/internal/types/tenant_skill.go b/internal/types/tenant_skill.go index 8841b42aad..fad685f964 100644 --- a/internal/types/tenant_skill.go +++ b/internal/types/tenant_skill.go @@ -16,7 +16,8 @@ const ( ) // Snapshot ledger states. deleted is written only after a real provider-side -// delete, which today happens only when the whole sandbox config is removed. +// delete: either the whole sandbox config is removed, or the reaper has +// pruned a retired snapshot past its retention window. const ( SkillSnapshotStateBuilding = "building" SkillSnapshotStateActive = "active"