fix(snapshot): make deletion retries durable
- commit
- 90ff4fa3a41e44b37b64cace87de56a7fe79d19e
- author
- brice · · 2026-07-22 00:50 UTC
- parent
- ae5758e
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
11 files · +286 -86
M internal/adapters/snapshot/legacy.go +121 -19
| @@ -13,12 +13,27 @@ import ( | |||
| 13 | 13 | "github.com/bnema/vev/internal/ports" | |
| 14 | 14 | ) | |
| 15 | 15 | ||
| 16 | + | const ( | |
| 17 | + | legacyDeleteMarkerPrefix = ".legacy-delete-" | |
| 18 | + | maxLegacyDeleteMarkers = maxLegacySnapshotFiles | |
| 19 | + | maxLegacyDeleteNameBytes = 200 | |
| 20 | + | ) | |
| 21 | + | ||
| 22 | + | type legacyDeleteMarker struct { | |
| 23 | + | name string | |
| 24 | + | } | |
| 25 | + | ||
| 16 | 26 | // LoadLegacy reads only pre-incremental root .snap files. It is deliberately | |
| 17 | 27 | // isolated from repository reads so incremental directories are never traversed. | |
| 18 | 28 | func (r *Repository) LoadLegacy(ctx context.Context) ([]ports.LegacySnapshot, error) { | |
| 19 | 29 | if err := ctx.Err(); err != nil { | |
| 20 | 30 | return nil, err | |
| 21 | 31 | } | |
| 32 | + | // A marker is an already-authorized deletion, not an import candidate. Finish | |
| 33 | + | // it before exposing any legacy blobs so process restart cannot republish one. | |
| 34 | + | if err := r.cleanupPendingLegacyDeletes(ctx); err != nil { | |
| 35 | + | return nil, err | |
| 36 | + | } | |
| 22 | 37 | f, err := os.Open(r.dir) | |
| 23 | 38 | if errors.Is(err, os.ErrNotExist) { | |
| 24 | 39 | return nil, nil | |
| @@ -83,10 +98,9 @@ func (r *Repository) LoadLegacy(ctx context.Context) ([]ports.LegacySnapshot, er | |||
| 83 | 98 | return out, nil | |
| 84 | 99 | } | |
| 85 | 100 | ||
| 86 | - | // DeleteLegacy deletes the deterministic v3 file and synchronizes the root. | |
| 87 | - | // Once an unlink succeeds, its directory sync remains pending until it succeeds; | |
| 88 | - | // this lets an absent-file retry complete the durability boundary without | |
| 89 | - | // deleting a file recreated in the meantime. | |
| 101 | + | // DeleteLegacy durably authorizes deletion before unlinking the legacy source. | |
| 102 | + | // The private marker survives process restart, so LoadLegacy retries an | |
| 103 | + | // authorized deletion rather than returning it for a second import. | |
| 90 | 104 | func (r *Repository) DeleteLegacy(ctx context.Context, name string) error { | |
| 91 | 105 | if err := ctx.Err(); err != nil { | |
| 92 | 106 | return err | |
| @@ -98,28 +112,116 @@ func (r *Repository) DeleteLegacy(ctx context.Context, name string) error { | |||
| 98 | 112 | if err := ctx.Err(); err != nil { | |
| 99 | 113 | return err | |
| 100 | 114 | } | |
| 101 | - | if _, pending := r.pendingLegacySync.Load(filename); pending { | |
| 102 | - | return r.syncPendingLegacy(ctx, filename) | |
| 115 | + | if err := r.authorizeLegacyDelete(name); err != nil { | |
| 116 | + | return err | |
| 103 | 117 | } | |
| 118 | + | return r.deleteAuthorizedLegacy(ctx, name, filename) | |
| 119 | + | } | |
| 104 | 120 | ||
| 105 | - | path := filepath.Join(r.dir, filename) | |
| 106 | - | if err := r.remove(path); errors.Is(err, os.ErrNotExist) { | |
| 107 | - | // Tombstones make a process-restart retry distinguish an already-unlinked | |
| 108 | - | // legacy source from a source that was absent before the kill. Re-sync the | |
| 109 | - | // root to complete the original unlink's durability boundary. | |
| 110 | - | killed, tombstoneErr := r.tombstoned(name) | |
| 111 | - | if tombstoneErr != nil { | |
| 112 | - | return fmt.Errorf("read killed session marker: %w", safeFilesystemError(tombstoneErr)) | |
| 121 | + | func legacyDeleteMarkerFilename(key string) string { return legacyDeleteMarkerPrefix + key } | |
| 122 | + | ||
| 123 | + | func (r *Repository) authorizeLegacyDelete(name string) error { | |
| 124 | + | if err := r.ensurePrivateDirectory(r.dir); err != nil { | |
| 125 | + | return fmt.Errorf("create legacy deletion marker directory: %w", safeFilesystemError(err)) | |
| 126 | + | } | |
| 127 | + | path := filepath.Join(r.dir, legacyDeleteMarkerFilename(sessionKey(name))) | |
| 128 | + | if err := r.writeImmutable(path, []byte(name), func(data []byte) error { | |
| 129 | + | if string(data) != name { | |
| 130 | + | return fmt.Errorf("legacy deletion marker identity mismatch") | |
| 113 | 131 | } | |
| 114 | - | if !killed { | |
| 115 | - | return nil | |
| 132 | + | return nil | |
| 133 | + | }); err != nil { | |
| 134 | + | return fmt.Errorf("write legacy deletion marker: %w", safeFilesystemError(err)) | |
| 135 | + | } | |
| 136 | + | return nil | |
| 137 | + | } | |
| 138 | + | ||
| 139 | + | func (r *Repository) deleteAuthorizedLegacy(ctx context.Context, name, filename string) error { | |
| 140 | + | if _, pending := r.pendingLegacySync.Load(filename); pending { | |
| 141 | + | if err := r.syncPendingLegacy(ctx, filename); err != nil { | |
| 142 | + | return err | |
| 116 | 143 | } | |
| 117 | - | return r.syncPendingLegacy(ctx, filename) | |
| 118 | - | } else if err != nil { | |
| 144 | + | return r.clearLegacyDeleteAuthorization(name) | |
| 145 | + | } | |
| 146 | + | path := filepath.Join(r.dir, filename) | |
| 147 | + | if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { | |
| 119 | 148 | return fmt.Errorf("delete legacy snapshot: %w", safeFilesystemError(err)) | |
| 120 | 149 | } | |
| 121 | 150 | r.pendingLegacySync.Store(filename, struct{}{}) | |
| 122 | - | return r.syncPendingLegacy(ctx, filename) | |
| 151 | + | if err := r.syncPendingLegacy(ctx, filename); err != nil { | |
| 152 | + | return err | |
| 153 | + | } | |
| 154 | + | return r.clearLegacyDeleteAuthorization(name) | |
| 155 | + | } | |
| 156 | + | ||
| 157 | + | func (r *Repository) clearLegacyDeleteAuthorization(name string) error { | |
| 158 | + | path := filepath.Join(r.dir, legacyDeleteMarkerFilename(sessionKey(name))) | |
| 159 | + | if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { | |
| 160 | + | return fmt.Errorf("clear legacy deletion marker: %w", safeFilesystemError(err)) | |
| 161 | + | } | |
| 162 | + | if err := r.syncDirectory(r.dir); err != nil { | |
| 163 | + | return fmt.Errorf("sync legacy deletion marker directory: %w", safeFilesystemError(err)) | |
| 164 | + | } | |
| 165 | + | return nil | |
| 166 | + | } | |
| 167 | + | ||
| 168 | + | func (r *Repository) cleanupPendingLegacyDeletes(ctx context.Context) error { | |
| 169 | + | markers, err := r.pendingLegacyDeleteMarkers(ctx) | |
| 170 | + | if errors.Is(err, os.ErrNotExist) { | |
| 171 | + | return nil | |
| 172 | + | } | |
| 173 | + | if err != nil { | |
| 174 | + | return err | |
| 175 | + | } | |
| 176 | + | for _, marker := range markers { | |
| 177 | + | if err := r.DeleteLegacy(ctx, marker.name); err != nil { | |
| 178 | + | return fmt.Errorf("retry authorized legacy deletion: %w", err) | |
| 179 | + | } | |
| 180 | + | } | |
| 181 | + | return nil | |
| 182 | + | } | |
| 183 | + | ||
| 184 | + | func (r *Repository) pendingLegacyDeleteMarkers(ctx context.Context) ([]legacyDeleteMarker, error) { | |
| 185 | + | f, err := os.Open(r.dir) | |
| 186 | + | if err != nil { | |
| 187 | + | return nil, err | |
| 188 | + | } | |
| 189 | + | defer f.Close() | |
| 190 | + | markers := make([]legacyDeleteMarker, 0, maxLegacyDeleteMarkers) | |
| 191 | + | for { | |
| 192 | + | entries, readErr := f.ReadDir(maintenanceBatch) | |
| 193 | + | for _, entry := range entries { | |
| 194 | + | if err := ctx.Err(); err != nil { | |
| 195 | + | return nil, err | |
| 196 | + | } | |
| 197 | + | if entry.IsDir() || !strings.HasPrefix(entry.Name(), legacyDeleteMarkerPrefix) { | |
| 198 | + | continue | |
| 199 | + | } | |
| 200 | + | if len(markers) == maxLegacyDeleteMarkers { | |
| 201 | + | return nil, fmt.Errorf("legacy deletion retry has too many markers (maximum %d)", maxLegacyDeleteMarkers) | |
| 202 | + | } | |
| 203 | + | key := strings.TrimPrefix(entry.Name(), legacyDeleteMarkerPrefix) | |
| 204 | + | if !canonicalSessionKey(key) { | |
| 205 | + | return nil, fmt.Errorf("invalid legacy deletion marker") | |
| 206 | + | } | |
| 207 | + | path := filepath.Join(r.dir, entry.Name()) | |
| 208 | + | if info, statErr := os.Lstat(path); statErr != nil || !info.Mode().IsRegular() || info.Size() > maxLegacyDeleteNameBytes { | |
| 209 | + | return nil, fmt.Errorf("invalid legacy deletion marker") | |
| 210 | + | } | |
| 211 | + | data, dataErr := readBounded(path) | |
| 212 | + | if dataErr != nil || len(data) > maxLegacyDeleteNameBytes || sessionKey(string(data)) != key { | |
| 213 | + | return nil, fmt.Errorf("invalid legacy deletion marker") | |
| 214 | + | } | |
| 215 | + | markers = append(markers, legacyDeleteMarker{name: string(data)}) | |
| 216 | + | } | |
| 217 | + | if errors.Is(readErr, io.EOF) { | |
| 218 | + | break | |
| 219 | + | } | |
| 220 | + | if readErr != nil { | |
| 221 | + | return nil, fmt.Errorf("read legacy deletion markers: %w", safeFilesystemError(readErr)) | |
| 222 | + | } | |
| 223 | + | } | |
| 224 | + | return markers, nil | |
| 123 | 225 | } | |
| 124 | 226 | ||
| 125 | 227 | // Tombstone durably marks name as logically killed before either snapshot | |
| @@ -13,12 +13,27 @@ import ( | |||||
| 13 | "github.com/bnema/vev/internal/ports" | 13 | "github.com/bnema/vev/internal/ports" | ||
| 14 | ) | 14 | ) | ||
| 15 | 15 | ||||
| 16 | + | const ( | |||
| 17 | + | legacyDeleteMarkerPrefix = ".legacy-delete-" | |||
| 18 | + | maxLegacyDeleteMarkers = maxLegacySnapshotFiles | |||
| 19 | + | maxLegacyDeleteNameBytes = 200 | |||
| 20 | + | ) | |||
| 21 | + | ||||
| 22 | + | type legacyDeleteMarker struct { | |||
| 23 | + | name string | |||
| 24 | + | } | |||
| 25 | + | ||||
| 16 | // LoadLegacy reads only pre-incremental root .snap files. It is deliberately | 26 | // LoadLegacy reads only pre-incremental root .snap files. It is deliberately | ||
| 17 | // isolated from repository reads so incremental directories are never traversed. | 27 | // isolated from repository reads so incremental directories are never traversed. | ||
| 18 | func (r *Repository) LoadLegacy(ctx context.Context) ([]ports.LegacySnapshot, error) { | 28 | func (r *Repository) LoadLegacy(ctx context.Context) ([]ports.LegacySnapshot, error) { | ||
| 19 | if err := ctx.Err(); err != nil { | 29 | if err := ctx.Err(); err != nil { | ||
| 20 | return nil, err | 30 | return nil, err | ||
| 21 | } | 31 | } | ||
| 32 | + | // A marker is an already-authorized deletion, not an import candidate. Finish | |||
| 33 | + | // it before exposing any legacy blobs so process restart cannot republish one. | |||
| 34 | + | if err := r.cleanupPendingLegacyDeletes(ctx); err != nil { | |||
| 35 | + | return nil, err | |||
| 36 | + | } | |||
| 22 | f, err := os.Open(r.dir) | 37 | f, err := os.Open(r.dir) | ||
| 23 | if errors.Is(err, os.ErrNotExist) { | 38 | if errors.Is(err, os.ErrNotExist) { | ||
| 24 | return nil, nil | 39 | return nil, nil | ||
| @@ -83,10 +98,9 @@ func (r *Repository) LoadLegacy(ctx context.Context) ([]ports.LegacySnapshot, er | |||||
| 83 | return out, nil | 98 | return out, nil | ||
| 84 | } | 99 | } | ||
| 85 | 100 | ||||
| 86 | - | // DeleteLegacy deletes the deterministic v3 file and synchronizes the root. | 101 | + | // DeleteLegacy durably authorizes deletion before unlinking the legacy source. |
| 87 | - | // Once an unlink succeeds, its directory sync remains pending until it succeeds; | 102 | + | // The private marker survives process restart, so LoadLegacy retries an |
| 88 | - | // this lets an absent-file retry complete the durability boundary without | 103 | + | // authorized deletion rather than returning it for a second import. |
| 89 | - | // deleting a file recreated in the meantime. | |||
| 90 | func (r *Repository) DeleteLegacy(ctx context.Context, name string) error { | 104 | func (r *Repository) DeleteLegacy(ctx context.Context, name string) error { | ||
| 91 | if err := ctx.Err(); err != nil { | 105 | if err := ctx.Err(); err != nil { | ||
| 92 | return err | 106 | return err | ||
| @@ -98,28 +112,116 @@ func (r *Repository) DeleteLegacy(ctx context.Context, name string) error { | |||||
| 98 | if err := ctx.Err(); err != nil { | 112 | if err := ctx.Err(); err != nil { | ||
| 99 | return err | 113 | return err | ||
| 100 | } | 114 | } | ||
| 101 | - | if _, pending := r.pendingLegacySync.Load(filename); pending { | 115 | + | if err := r.authorizeLegacyDelete(name); err != nil { |
| 102 | - | return r.syncPendingLegacy(ctx, filename) | 116 | + | return err |
| 103 | } | 117 | } | ||
| 118 | + | return r.deleteAuthorizedLegacy(ctx, name, filename) | |||
| 119 | + | } | |||
| 104 | 120 | ||||
| 105 | - | path := filepath.Join(r.dir, filename) | 121 | + | func legacyDeleteMarkerFilename(key string) string { return legacyDeleteMarkerPrefix + key } |
| 106 | - | if err := r.remove(path); errors.Is(err, os.ErrNotExist) { | 122 | + | |
| 107 | - | // Tombstones make a process-restart retry distinguish an already-unlinked | 123 | + | func (r *Repository) authorizeLegacyDelete(name string) error { |
| 108 | - | // legacy source from a source that was absent before the kill. Re-sync the | 124 | + | if err := r.ensurePrivateDirectory(r.dir); err != nil { |
| 109 | - | // root to complete the original unlink's durability boundary. | 125 | + | return fmt.Errorf("create legacy deletion marker directory: %w", safeFilesystemError(err)) |
| 110 | - | killed, tombstoneErr := r.tombstoned(name) | 126 | + | } |
| 111 | - | if tombstoneErr != nil { | 127 | + | path := filepath.Join(r.dir, legacyDeleteMarkerFilename(sessionKey(name))) |
| 112 | - | return fmt.Errorf("read killed session marker: %w", safeFilesystemError(tombstoneErr)) | 128 | + | if err := r.writeImmutable(path, []byte(name), func(data []byte) error { |
| 129 | + | if string(data) != name { | |||
| 130 | + | return fmt.Errorf("legacy deletion marker identity mismatch") | |||
| 113 | } | 131 | } | ||
| 114 | - | if !killed { | 132 | + | return nil |
| 115 | - | return nil | 133 | + | }); err != nil { |
| 134 | + | return fmt.Errorf("write legacy deletion marker: %w", safeFilesystemError(err)) | |||
| 135 | + | } | |||
| 136 | + | return nil | |||
| 137 | + | } | |||
| 138 | + | ||||
| 139 | + | func (r *Repository) deleteAuthorizedLegacy(ctx context.Context, name, filename string) error { | |||
| 140 | + | if _, pending := r.pendingLegacySync.Load(filename); pending { | |||
| 141 | + | if err := r.syncPendingLegacy(ctx, filename); err != nil { | |||
| 142 | + | return err | |||
| 116 | } | 143 | } | ||
| 117 | - | return r.syncPendingLegacy(ctx, filename) | 144 | + | return r.clearLegacyDeleteAuthorization(name) |
| 118 | - | } else if err != nil { | 145 | + | } |
| 146 | + | path := filepath.Join(r.dir, filename) | |||
| 147 | + | if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { | |||
| 119 | return fmt.Errorf("delete legacy snapshot: %w", safeFilesystemError(err)) | 148 | return fmt.Errorf("delete legacy snapshot: %w", safeFilesystemError(err)) | ||
| 120 | } | 149 | } | ||
| 121 | r.pendingLegacySync.Store(filename, struct{}{}) | 150 | r.pendingLegacySync.Store(filename, struct{}{}) | ||
| 122 | - | return r.syncPendingLegacy(ctx, filename) | 151 | + | if err := r.syncPendingLegacy(ctx, filename); err != nil { |
| 152 | + | return err | |||
| 153 | + | } | |||
| 154 | + | return r.clearLegacyDeleteAuthorization(name) | |||
| 155 | + | } | |||
| 156 | + | ||||
| 157 | + | func (r *Repository) clearLegacyDeleteAuthorization(name string) error { | |||
| 158 | + | path := filepath.Join(r.dir, legacyDeleteMarkerFilename(sessionKey(name))) | |||
| 159 | + | if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { | |||
| 160 | + | return fmt.Errorf("clear legacy deletion marker: %w", safeFilesystemError(err)) | |||
| 161 | + | } | |||
| 162 | + | if err := r.syncDirectory(r.dir); err != nil { | |||
| 163 | + | return fmt.Errorf("sync legacy deletion marker directory: %w", safeFilesystemError(err)) | |||
| 164 | + | } | |||
| 165 | + | return nil | |||
| 166 | + | } | |||
| 167 | + | ||||
| 168 | + | func (r *Repository) cleanupPendingLegacyDeletes(ctx context.Context) error { | |||
| 169 | + | markers, err := r.pendingLegacyDeleteMarkers(ctx) | |||
| 170 | + | if errors.Is(err, os.ErrNotExist) { | |||
| 171 | + | return nil | |||
| 172 | + | } | |||
| 173 | + | if err != nil { | |||
| 174 | + | return err | |||
| 175 | + | } | |||
| 176 | + | for _, marker := range markers { | |||
| 177 | + | if err := r.DeleteLegacy(ctx, marker.name); err != nil { | |||
| 178 | + | return fmt.Errorf("retry authorized legacy deletion: %w", err) | |||
| 179 | + | } | |||
| 180 | + | } | |||
| 181 | + | return nil | |||
| 182 | + | } | |||
| 183 | + | ||||
| 184 | + | func (r *Repository) pendingLegacyDeleteMarkers(ctx context.Context) ([]legacyDeleteMarker, error) { | |||
| 185 | + | f, err := os.Open(r.dir) | |||
| 186 | + | if err != nil { | |||
| 187 | + | return nil, err | |||
| 188 | + | } | |||
| 189 | + | defer f.Close() | |||
| 190 | + | markers := make([]legacyDeleteMarker, 0, maxLegacyDeleteMarkers) | |||
| 191 | + | for { | |||
| 192 | + | entries, readErr := f.ReadDir(maintenanceBatch) | |||
| 193 | + | for _, entry := range entries { | |||
| 194 | + | if err := ctx.Err(); err != nil { | |||
| 195 | + | return nil, err | |||
| 196 | + | } | |||
| 197 | + | if entry.IsDir() || !strings.HasPrefix(entry.Name(), legacyDeleteMarkerPrefix) { | |||
| 198 | + | continue | |||
| 199 | + | } | |||
| 200 | + | if len(markers) == maxLegacyDeleteMarkers { | |||
| 201 | + | return nil, fmt.Errorf("legacy deletion retry has too many markers (maximum %d)", maxLegacyDeleteMarkers) | |||
| 202 | + | } | |||
| 203 | + | key := strings.TrimPrefix(entry.Name(), legacyDeleteMarkerPrefix) | |||
| 204 | + | if !canonicalSessionKey(key) { | |||
| 205 | + | return nil, fmt.Errorf("invalid legacy deletion marker") | |||
| 206 | + | } | |||
| 207 | + | path := filepath.Join(r.dir, entry.Name()) | |||
| 208 | + | if info, statErr := os.Lstat(path); statErr != nil || !info.Mode().IsRegular() || info.Size() > maxLegacyDeleteNameBytes { | |||
| 209 | + | return nil, fmt.Errorf("invalid legacy deletion marker") | |||
| 210 | + | } | |||
| 211 | + | data, dataErr := readBounded(path) | |||
| 212 | + | if dataErr != nil || len(data) > maxLegacyDeleteNameBytes || sessionKey(string(data)) != key { | |||
| 213 | + | return nil, fmt.Errorf("invalid legacy deletion marker") | |||
| 214 | + | } | |||
| 215 | + | markers = append(markers, legacyDeleteMarker{name: string(data)}) | |||
| 216 | + | } | |||
| 217 | + | if errors.Is(readErr, io.EOF) { | |||
| 218 | + | break | |||
| 219 | + | } | |||
| 220 | + | if readErr != nil { | |||
| 221 | + | return nil, fmt.Errorf("read legacy deletion markers: %w", safeFilesystemError(readErr)) | |||
| 222 | + | } | |||
| 223 | + | } | |||
| 224 | + | return markers, nil | |||
| 123 | } | 225 | } | ||
| 124 | 226 | ||||
| 125 | // Tombstone durably marks name as logically killed before either snapshot | 227 | // Tombstone durably marks name as logically killed before either snapshot | ||
A internal/adapters/snapshot/legacy_pending_delete_test.go +36 -0
| @@ -0,0 +1,36 @@ | |||
| 1 | + | package snapshot | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | "errors" | |
| 6 | + | "os" | |
| 7 | + | "path/filepath" | |
| 8 | + | "testing" | |
| 9 | + | ||
| 10 | + | "github.com/stretchr/testify/require" | |
| 11 | + | ) | |
| 12 | + | ||
| 13 | + | func TestRepositoryLoadLegacyCompletesAuthorizedDeleteAfterRestart(t *testing.T) { | |
| 14 | + | repo := NewRepository(privateDir(t)) | |
| 15 | + | require.NoError(t, os.Mkdir(repo.dir, 0o700)) | |
| 16 | + | path := filepath.Join(repo.dir, filenameForName("named")) | |
| 17 | + | require.NoError(t, os.WriteFile(path, []byte("legacy"), 0o600)) | |
| 18 | + | ||
| 19 | + | injected := errors.New("unlink failed") | |
| 20 | + | repo.hooks.remove = func(path string) error { | |
| 21 | + | if filepath.Base(path) == filenameForName("named") { | |
| 22 | + | return injected | |
| 23 | + | } | |
| 24 | + | return os.Remove(path) | |
| 25 | + | } | |
| 26 | + | require.ErrorIs(t, repo.DeleteLegacy(context.Background(), "named"), injected) | |
| 27 | + | ||
| 28 | + | // A new repository has no in-memory retry state. The durable authorization | |
| 29 | + | // must fence import and complete deletion without another publication. | |
| 30 | + | repo = NewRepository(repo.dir) | |
| 31 | + | legacy, err := repo.LoadLegacy(context.Background()) | |
| 32 | + | require.NoError(t, err) | |
| 33 | + | require.Empty(t, legacy) | |
| 34 | + | _, err = os.Stat(path) | |
| 35 | + | require.ErrorIs(t, err, os.ErrNotExist) | |
| 36 | + | } | |
| @@ -0,0 +1,36 @@ | |||||
| 1 | + | package snapshot | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "context" | |||
| 5 | + | "errors" | |||
| 6 | + | "os" | |||
| 7 | + | "path/filepath" | |||
| 8 | + | "testing" | |||
| 9 | + | ||||
| 10 | + | "github.com/stretchr/testify/require" | |||
| 11 | + | ) | |||
| 12 | + | ||||
| 13 | + | func TestRepositoryLoadLegacyCompletesAuthorizedDeleteAfterRestart(t *testing.T) { | |||
| 14 | + | repo := NewRepository(privateDir(t)) | |||
| 15 | + | require.NoError(t, os.Mkdir(repo.dir, 0o700)) | |||
| 16 | + | path := filepath.Join(repo.dir, filenameForName("named")) | |||
| 17 | + | require.NoError(t, os.WriteFile(path, []byte("legacy"), 0o600)) | |||
| 18 | + | ||||
| 19 | + | injected := errors.New("unlink failed") | |||
| 20 | + | repo.hooks.remove = func(path string) error { | |||
| 21 | + | if filepath.Base(path) == filenameForName("named") { | |||
| 22 | + | return injected | |||
| 23 | + | } | |||
| 24 | + | return os.Remove(path) | |||
| 25 | + | } | |||
| 26 | + | require.ErrorIs(t, repo.DeleteLegacy(context.Background(), "named"), injected) | |||
| 27 | + | ||||
| 28 | + | // A new repository has no in-memory retry state. The durable authorization | |||
| 29 | + | // must fence import and complete deletion without another publication. | |||
| 30 | + | repo = NewRepository(repo.dir) | |||
| 31 | + | legacy, err := repo.LoadLegacy(context.Background()) | |||
| 32 | + | require.NoError(t, err) | |||
| 33 | + | require.Empty(t, legacy) | |||
| 34 | + | _, err = os.Stat(path) | |||
| 35 | + | require.ErrorIs(t, err, os.ErrNotExist) | |||
| 36 | + | } | |||
M internal/adapters/snapshot/legacy_test.go +1 -1
| @@ -95,7 +95,7 @@ func TestRepositoryDeleteLegacySurfacesRootSyncAndCanRetry(t *testing.T) { | |||
| 95 | 95 | if err := repo.DeleteLegacy(context.Background(), "named"); err != nil { | |
| 96 | 96 | t.Fatalf("retry DeleteLegacy: %v", err) | |
| 97 | 97 | } | |
| 98 | - | if syncCalls != 2 { | |
| 98 | + | if syncCalls != 4 { | |
| 99 | 99 | t.Fatalf("sync calls = %d, want retry to sync absent file", syncCalls) | |
| 100 | 100 | } | |
| 101 | 101 | } | |
| @@ -95,7 +95,7 @@ func TestRepositoryDeleteLegacySurfacesRootSyncAndCanRetry(t *testing.T) { | |||||
| 95 | if err := repo.DeleteLegacy(context.Background(), "named"); err != nil { | 95 | if err := repo.DeleteLegacy(context.Background(), "named"); err != nil { | ||
| 96 | t.Fatalf("retry DeleteLegacy: %v", err) | 96 | t.Fatalf("retry DeleteLegacy: %v", err) | ||
| 97 | } | 97 | } | ||
| 98 | - | if syncCalls != 2 { | 98 | + | if syncCalls != 4 { |
| 99 | t.Fatalf("sync calls = %d, want retry to sync absent file", syncCalls) | 99 | t.Fatalf("sync calls = %d, want retry to sync absent file", syncCalls) | ||
| 100 | } | 100 | } | ||
| 101 | } | 101 | } | ||
M internal/adapters/snapshot/offline_kill_tombstone_test.go +1 -1
| @@ -49,5 +49,5 @@ func TestRepositoryDeleteLegacyRetriesDurabilityAfterTombstone(t *testing.T) { | |||
| 49 | 49 | return nil | |
| 50 | 50 | } | |
| 51 | 51 | require.NoError(t, repo.DeleteLegacy(context.Background(), "named")) | |
| 52 | - | require.Equal(t, 2, calls, "a tombstoned missing legacy source must resync its unlink") | |
| 52 | + | require.Equal(t, 4, calls, "authorization and its cleanup are durably synced around the retry") | |
| 53 | 53 | } | |
| @@ -49,5 +49,5 @@ func TestRepositoryDeleteLegacyRetriesDurabilityAfterTombstone(t *testing.T) { | |||||
| 49 | return nil | 49 | return nil | ||
| 50 | } | 50 | } | ||
| 51 | require.NoError(t, repo.DeleteLegacy(context.Background(), "named")) | 51 | require.NoError(t, repo.DeleteLegacy(context.Background(), "named")) | ||
| 52 | - | require.Equal(t, 2, calls, "a tombstoned missing legacy source must resync its unlink") | 52 | + | require.Equal(t, 4, calls, "authorization and its cleanup are durably synced around the retry") |
| 53 | } | 53 | } | ||
M internal/app/run.go +4 -4
| @@ -1253,14 +1253,14 @@ func runKill(ctx context.Context, name string, all, daemon bool) error { | |||
| 1253 | 1253 | phaseError("deleting legacy session snapshot", legacyErr), | |
| 1254 | 1254 | ) | |
| 1255 | 1255 | } | |
| 1256 | - | if tombstoneErr := snapshots.DeleteTombstone(ctx, name); tombstoneErr != nil { | |
| 1257 | - | _ = p.Close() | |
| 1258 | - | return fmt.Errorf("vev: clearing killed session marker: %w", tombstoneErr) | |
| 1259 | - | } | |
| 1260 | 1256 | if deleteErr := p.Delete(name); deleteErr != nil { | |
| 1261 | 1257 | _ = p.Close() | |
| 1262 | 1258 | return fmt.Errorf("vev: deleting stored session: %w", deleteErr) | |
| 1263 | 1259 | } | |
| 1260 | + | if tombstoneErr := snapshots.DeleteTombstone(ctx, name); tombstoneErr != nil { | |
| 1261 | + | _ = p.Close() | |
| 1262 | + | return fmt.Errorf("vev: clearing killed session marker: %w", tombstoneErr) | |
| 1263 | + | } | |
| 1264 | 1264 | if closeErr := p.Close(); closeErr != nil { | |
| 1265 | 1265 | return fmt.Errorf("vev: closing stored sessions: %w", closeErr) | |
| 1266 | 1266 | } | |
| @@ -1253,14 +1253,14 @@ func runKill(ctx context.Context, name string, all, daemon bool) error { | |||||
| 1253 | phaseError("deleting legacy session snapshot", legacyErr), | 1253 | phaseError("deleting legacy session snapshot", legacyErr), | ||
| 1254 | ) | 1254 | ) | ||
| 1255 | } | 1255 | } | ||
| 1256 | - | if tombstoneErr := snapshots.DeleteTombstone(ctx, name); tombstoneErr != nil { | |||
| 1257 | - | _ = p.Close() | |||
| 1258 | - | return fmt.Errorf("vev: clearing killed session marker: %w", tombstoneErr) | |||
| 1259 | - | } | |||
| 1260 | if deleteErr := p.Delete(name); deleteErr != nil { | 1256 | if deleteErr := p.Delete(name); deleteErr != nil { | ||
| 1261 | _ = p.Close() | 1257 | _ = p.Close() | ||
| 1262 | return fmt.Errorf("vev: deleting stored session: %w", deleteErr) | 1258 | return fmt.Errorf("vev: deleting stored session: %w", deleteErr) | ||
| 1263 | } | 1259 | } | ||
| 1260 | + | if tombstoneErr := snapshots.DeleteTombstone(ctx, name); tombstoneErr != nil { | |||
| 1261 | + | _ = p.Close() | |||
| 1262 | + | return fmt.Errorf("vev: clearing killed session marker: %w", tombstoneErr) | |||
| 1263 | + | } | |||
| 1264 | if closeErr := p.Close(); closeErr != nil { | 1264 | if closeErr := p.Close(); closeErr != nil { | ||
| 1265 | return fmt.Errorf("vev: closing stored sessions: %w", closeErr) | 1265 | return fmt.Errorf("vev: closing stored sessions: %w", closeErr) | ||
| 1266 | } | 1266 | } | ||
M internal/usecase/daemon/daemon.go +7 -31
| @@ -7,7 +7,6 @@ import ( | |||
| 7 | 7 | "log/slog" | |
| 8 | 8 | "math" | |
| 9 | 9 | "os" | |
| 10 | - | "slices" | |
| 11 | 10 | "sort" | |
| 12 | 11 | "sync" | |
| 13 | 12 | "sync/atomic" | |
| @@ -219,15 +218,6 @@ type stoppedSession struct { | |||
| 219 | 218 | purging bool | |
| 220 | 219 | } | |
| 221 | 220 | ||
| 222 | - | func (s stoppedSession) same(other stoppedSession) bool { | |
| 223 | - | return s.name == other.name && | |
| 224 | - | s.cwd == other.cwd && | |
| 225 | - | s.createdAt == other.createdAt && | |
| 226 | - | s.lastUsedSeq == other.lastUsedSeq && | |
| 227 | - | s.purging == other.purging && | |
| 228 | - | slices.Equal(s.tabNames, other.tabNames) | |
| 229 | - | } | |
| 230 | - | ||
| 231 | 221 | type Option func(*Daemon) | |
| 232 | 222 | ||
| 233 | 223 | // WithRuntimeObserver accepts only a composition-root serialized observer. | |
| @@ -712,29 +702,15 @@ func (d *Daemon) handleKill(tr ports.Transport, f ports.Frame) { | |||
| 712 | 702 | d.mu.Lock() | |
| 713 | 703 | target := d.findByNameLocked(k.Name) | |
| 714 | 704 | if target == nil { | |
| 715 | - | if stopped, ok := d.stopped[k.Name]; ok { | |
| 705 | + | if _, ok := d.stopped[k.Name]; ok { | |
| 716 | 706 | d.mu.Unlock() | |
| 717 | - | if d.snapsEnabled { | |
| 718 | - | var deleteErr error | |
| 719 | - | if d.snapshotRepository != nil { | |
| 720 | - | deleteErr = d.snapshotRepository.Delete(context.Background(), k.Name) | |
| 721 | - | } | |
| 722 | - | if deleteErr != nil { | |
| 723 | - | d.log.Warn("deleting stopped session snapshot failed", "err", deleteErr, "session", k.Name) | |
| 724 | - | _ = tr.Send(frameError(ports.ErrInternal, "deleting stopped session snapshot failed")) | |
| 725 | - | return | |
| 726 | - | } | |
| 727 | - | } | |
| 728 | - | if err := d.persist.Delete(k.Name); err != nil { | |
| 729 | - | d.log.Warn("deleting persisted stopped session failed", "err", err, "session", k.Name) | |
| 730 | - | _ = tr.Send(frameError(ports.ErrInternal, "deleting persisted stopped session failed")) | |
| 731 | - | return | |
| 707 | + | // Stopped sessions follow the same tombstone-backed transaction as | |
| 708 | + | // live and offline purges. In particular, a retained legacy source | |
| 709 | + | // must never outlive metadata after an interrupted control kill. | |
| 710 | + | if err := d.retryStoppedPurge(k.Name); err != nil { | |
| 711 | + | d.log.Warn("deleting stopped session failed", "err", err, "session", k.Name) | |
| 712 | + | _ = tr.Send(frameError(ports.ErrInternal, "deleting stopped session failed")) | |
| 732 | 713 | } | |
| 733 | - | d.mu.Lock() | |
| 734 | - | if cur, ok := d.stopped[k.Name]; ok && cur.same(stopped) { | |
| 735 | - | delete(d.stopped, k.Name) | |
| 736 | - | } | |
| 737 | - | d.mu.Unlock() | |
| 738 | 714 | return | |
| 739 | 715 | } | |
| 740 | 716 | } | |
| @@ -7,7 +7,6 @@ import ( | |||||
| 7 | "log/slog" | 7 | "log/slog" | ||
| 8 | "math" | 8 | "math" | ||
| 9 | "os" | 9 | "os" | ||
| 10 | - | "slices" | |||
| 11 | "sort" | 10 | "sort" | ||
| 12 | "sync" | 11 | "sync" | ||
| 13 | "sync/atomic" | 12 | "sync/atomic" | ||
| @@ -219,15 +218,6 @@ type stoppedSession struct { | |||||
| 219 | purging bool | 218 | purging bool | ||
| 220 | } | 219 | } | ||
| 221 | 220 | ||||
| 222 | - | func (s stoppedSession) same(other stoppedSession) bool { | |||
| 223 | - | return s.name == other.name && | |||
| 224 | - | s.cwd == other.cwd && | |||
| 225 | - | s.createdAt == other.createdAt && | |||
| 226 | - | s.lastUsedSeq == other.lastUsedSeq && | |||
| 227 | - | s.purging == other.purging && | |||
| 228 | - | slices.Equal(s.tabNames, other.tabNames) | |||
| 229 | - | } | |||
| 230 | - | ||||
| 231 | type Option func(*Daemon) | 221 | type Option func(*Daemon) | ||
| 232 | 222 | ||||
| 233 | // WithRuntimeObserver accepts only a composition-root serialized observer. | 223 | // WithRuntimeObserver accepts only a composition-root serialized observer. | ||
| @@ -712,29 +702,15 @@ func (d *Daemon) handleKill(tr ports.Transport, f ports.Frame) { | |||||
| 712 | d.mu.Lock() | 702 | d.mu.Lock() | ||
| 713 | target := d.findByNameLocked(k.Name) | 703 | target := d.findByNameLocked(k.Name) | ||
| 714 | if target == nil { | 704 | if target == nil { | ||
| 715 | - | if stopped, ok := d.stopped[k.Name]; ok { | 705 | + | if _, ok := d.stopped[k.Name]; ok { |
| 716 | d.mu.Unlock() | 706 | d.mu.Unlock() | ||
| 717 | - | if d.snapsEnabled { | 707 | + | // Stopped sessions follow the same tombstone-backed transaction as |
| 718 | - | var deleteErr error | 708 | + | // live and offline purges. In particular, a retained legacy source |
| 719 | - | if d.snapshotRepository != nil { | 709 | + | // must never outlive metadata after an interrupted control kill. |
| 720 | - | deleteErr = d.snapshotRepository.Delete(context.Background(), k.Name) | 710 | + | if err := d.retryStoppedPurge(k.Name); err != nil { |
| 721 | - | } | 711 | + | d.log.Warn("deleting stopped session failed", "err", err, "session", k.Name) |
| 722 | - | if deleteErr != nil { | 712 | + | _ = tr.Send(frameError(ports.ErrInternal, "deleting stopped session failed")) |
| 723 | - | d.log.Warn("deleting stopped session snapshot failed", "err", deleteErr, "session", k.Name) | |||
| 724 | - | _ = tr.Send(frameError(ports.ErrInternal, "deleting stopped session snapshot failed")) | |||
| 725 | - | return | |||
| 726 | - | } | |||
| 727 | - | } | |||
| 728 | - | if err := d.persist.Delete(k.Name); err != nil { | |||
| 729 | - | d.log.Warn("deleting persisted stopped session failed", "err", err, "session", k.Name) | |||
| 730 | - | _ = tr.Send(frameError(ports.ErrInternal, "deleting persisted stopped session failed")) | |||
| 731 | - | return | |||
| 732 | } | 713 | } | ||
| 733 | - | d.mu.Lock() | |||
| 734 | - | if cur, ok := d.stopped[k.Name]; ok && cur.same(stopped) { | |||
| 735 | - | delete(d.stopped, k.Name) | |||
| 736 | - | } | |||
| 737 | - | d.mu.Unlock() | |||
| 738 | return | 714 | return | ||
| 739 | } | 715 | } | ||
| 740 | } | 716 | } | ||
M internal/usecase/daemon/session.go +18 -26
| @@ -556,9 +556,9 @@ func (d *Daemon) renameSession(sess *session, name string) error { | |||
| 556 | 556 | } | |
| 557 | 557 | ||
| 558 | 558 | // Reserve the rename while state is inspected, then perform persistence and | |
| 559 | - | // repository I/O without daemon/session locks. In particular, Delete must | |
| 560 | - | // follow coordinator quarantine+join or a late Publish could recreate the | |
| 561 | - | // old repository identity. | |
| 559 | + | // repository I/O without daemon/session locks. The old identity is first | |
| 560 | + | // quarantined and durably purged; committing the new record before that | |
| 561 | + | // transaction completes could leave a restorable old legacy source. | |
| 562 | 562 | d.mu.Lock() | |
| 563 | 563 | if taken := d.findByNameLocked(name); taken != nil && taken != sess { | |
| 564 | 564 | d.mu.Unlock() | |
| @@ -597,36 +597,30 @@ func (d *Daemon) renameSession(sess *session, name string) error { | |||
| 597 | 597 | d.mu.Unlock() | |
| 598 | 598 | ||
| 599 | 599 | rollback := func(err error) error { | |
| 600 | - | // The old record remains authoritative until its deletion succeeds. If | |
| 601 | - | // repository deletion was attempted, its outcome is not safely | |
| 602 | - | // reversible, so the coordinator stays quarantined rather than risking a | |
| 603 | - | // publication that recreates the deleted old identity. | |
| 604 | - | if cleanupErr := d.persist.Delete(name); cleanupErr != nil { | |
| 605 | - | d.log.Warn("cleaning up renamed persisted session failed", "err", cleanupErr, "session", name) | |
| 606 | - | } | |
| 607 | 600 | sess.mu.Lock() | |
| 608 | 601 | sess.renameInProgress = false | |
| 609 | 602 | sess.mu.Unlock() | |
| 610 | 603 | return err | |
| 611 | 604 | } | |
| 612 | - | if wasEphemeral || oldName != name { | |
| 613 | - | if err := d.persist.Save(record); err != nil { | |
| 614 | - | sess.mu.Lock() | |
| 615 | - | sess.renameInProgress = false | |
| 616 | - | sess.mu.Unlock() | |
| 617 | - | return err | |
| 618 | - | } | |
| 619 | - | } | |
| 620 | 605 | var quarantine snapshotCoordinatorQuarantine | |
| 621 | 606 | if !wasEphemeral && oldName != name { | |
| 622 | 607 | quarantine = quarantineSnapshotCoordinatorWithEpoch(sess) | |
| 623 | 608 | <-quarantine.done | |
| 624 | - | if d.snapsEnabled && d.snapshotRepository != nil { | |
| 625 | - | if err := d.snapshotRepository.Delete(context.Background(), oldName); err != nil { | |
| 626 | - | return rollback(err) | |
| 627 | - | } | |
| 609 | + | if err := d.beginSnapshotPurge(oldName); err != nil { | |
| 610 | + | return rollback(err) | |
| 611 | + | } | |
| 612 | + | // finishSnapshotPurge preserves the tombstone through both source | |
| 613 | + | // deletions and the old metadata delete, including a retained legacy | |
| 614 | + | // import whose verified source delete is still pending. | |
| 615 | + | if err := d.finishSnapshotPurge(oldName); err != nil { | |
| 616 | + | return rollback(err) | |
| 628 | 617 | } | |
| 629 | - | if err := d.persist.Delete(oldName); err != nil { | |
| 618 | + | } | |
| 619 | + | // For a named rename this is the durable commit point, deliberately after | |
| 620 | + | // the old tombstone transaction. Ephemeral promotion has no old durable | |
| 621 | + | // identity and can create its first record directly. | |
| 622 | + | if wasEphemeral || oldName != name { | |
| 623 | + | if err := d.persist.Save(record); err != nil { | |
| 630 | 624 | return rollback(err) | |
| 631 | 625 | } | |
| 632 | 626 | } | |
| @@ -854,9 +848,7 @@ func (d *Daemon) retryStoppedPurge(name string) error { | |||
| 854 | 848 | return err | |
| 855 | 849 | } | |
| 856 | 850 | d.mu.Lock() | |
| 857 | - | if _, ok := d.stopped[name]; ok { | |
| 858 | - | delete(d.stopped, name) | |
| 859 | - | } | |
| 851 | + | delete(d.stopped, name) | |
| 860 | 852 | d.mu.Unlock() | |
| 861 | 853 | return nil | |
| 862 | 854 | } | |
| @@ -556,9 +556,9 @@ func (d *Daemon) renameSession(sess *session, name string) error { | |||||
| 556 | } | 556 | } | ||
| 557 | 557 | ||||
| 558 | // Reserve the rename while state is inspected, then perform persistence and | 558 | // Reserve the rename while state is inspected, then perform persistence and | ||
| 559 | - | // repository I/O without daemon/session locks. In particular, Delete must | 559 | + | // repository I/O without daemon/session locks. The old identity is first |
| 560 | - | // follow coordinator quarantine+join or a late Publish could recreate the | 560 | + | // quarantined and durably purged; committing the new record before that |
| 561 | - | // old repository identity. | 561 | + | // transaction completes could leave a restorable old legacy source. |
| 562 | d.mu.Lock() | 562 | d.mu.Lock() | ||
| 563 | if taken := d.findByNameLocked(name); taken != nil && taken != sess { | 563 | if taken := d.findByNameLocked(name); taken != nil && taken != sess { | ||
| 564 | d.mu.Unlock() | 564 | d.mu.Unlock() | ||
| @@ -597,36 +597,30 @@ func (d *Daemon) renameSession(sess *session, name string) error { | |||||
| 597 | d.mu.Unlock() | 597 | d.mu.Unlock() | ||
| 598 | 598 | ||||
| 599 | rollback := func(err error) error { | 599 | rollback := func(err error) error { | ||
| 600 | - | // The old record remains authoritative until its deletion succeeds. If | |||
| 601 | - | // repository deletion was attempted, its outcome is not safely | |||
| 602 | - | // reversible, so the coordinator stays quarantined rather than risking a | |||
| 603 | - | // publication that recreates the deleted old identity. | |||
| 604 | - | if cleanupErr := d.persist.Delete(name); cleanupErr != nil { | |||
| 605 | - | d.log.Warn("cleaning up renamed persisted session failed", "err", cleanupErr, "session", name) | |||
| 606 | - | } | |||
| 607 | sess.mu.Lock() | 600 | sess.mu.Lock() | ||
| 608 | sess.renameInProgress = false | 601 | sess.renameInProgress = false | ||
| 609 | sess.mu.Unlock() | 602 | sess.mu.Unlock() | ||
| 610 | return err | 603 | return err | ||
| 611 | } | 604 | } | ||
| 612 | - | if wasEphemeral || oldName != name { | |||
| 613 | - | if err := d.persist.Save(record); err != nil { | |||
| 614 | - | sess.mu.Lock() | |||
| 615 | - | sess.renameInProgress = false | |||
| 616 | - | sess.mu.Unlock() | |||
| 617 | - | return err | |||
| 618 | - | } | |||
| 619 | - | } | |||
| 620 | var quarantine snapshotCoordinatorQuarantine | 605 | var quarantine snapshotCoordinatorQuarantine | ||
| 621 | if !wasEphemeral && oldName != name { | 606 | if !wasEphemeral && oldName != name { | ||
| 622 | quarantine = quarantineSnapshotCoordinatorWithEpoch(sess) | 607 | quarantine = quarantineSnapshotCoordinatorWithEpoch(sess) | ||
| 623 | <-quarantine.done | 608 | <-quarantine.done | ||
| 624 | - | if d.snapsEnabled && d.snapshotRepository != nil { | 609 | + | if err := d.beginSnapshotPurge(oldName); err != nil { |
| 625 | - | if err := d.snapshotRepository.Delete(context.Background(), oldName); err != nil { | 610 | + | return rollback(err) |
| 626 | - | return rollback(err) | 611 | + | } |
| 627 | - | } | 612 | + | // finishSnapshotPurge preserves the tombstone through both source |
| 613 | + | // deletions and the old metadata delete, including a retained legacy | |||
| 614 | + | // import whose verified source delete is still pending. | |||
| 615 | + | if err := d.finishSnapshotPurge(oldName); err != nil { | |||
| 616 | + | return rollback(err) | |||
| 628 | } | 617 | } | ||
| 629 | - | if err := d.persist.Delete(oldName); err != nil { | 618 | + | } |
| 619 | + | // For a named rename this is the durable commit point, deliberately after | |||
| 620 | + | // the old tombstone transaction. Ephemeral promotion has no old durable | |||
| 621 | + | // identity and can create its first record directly. | |||
| 622 | + | if wasEphemeral || oldName != name { | |||
| 623 | + | if err := d.persist.Save(record); err != nil { | |||
| 630 | return rollback(err) | 624 | return rollback(err) | ||
| 631 | } | 625 | } | ||
| 632 | } | 626 | } | ||
| @@ -854,9 +848,7 @@ func (d *Daemon) retryStoppedPurge(name string) error { | |||||
| 854 | return err | 848 | return err | ||
| 855 | } | 849 | } | ||
| 856 | d.mu.Lock() | 850 | d.mu.Lock() | ||
| 857 | - | if _, ok := d.stopped[name]; ok { | 851 | + | delete(d.stopped, name) |
| 858 | - | delete(d.stopped, name) | |||
| 859 | - | } | |||
| 860 | d.mu.Unlock() | 852 | d.mu.Unlock() | ||
| 861 | return nil | 853 | return nil | ||
| 862 | } | 854 | } | ||
A internal/usecase/daemon/session_purge_control_test.go +51 -0
| @@ -0,0 +1,51 @@ | |||
| 1 | + | package daemon | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "errors" | |
| 5 | + | "testing" | |
| 6 | + | ||
| 7 | + | "github.com/stretchr/testify/mock" | |
| 8 | + | "github.com/stretchr/testify/require" | |
| 9 | + | ||
| 10 | + | "github.com/bnema/vev/internal/ports" | |
| 11 | + | portsmocks "github.com/bnema/vev/internal/ports/mocks" | |
| 12 | + | ) | |
| 13 | + | ||
| 14 | + | func TestHandleKillStoppedUsesDurableBothSourcePurge(t *testing.T) { | |
| 15 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | |
| 16 | + | repository := &retryablePurgeRepository{legacyErr: errors.New("legacy delete failed")} | |
| 17 | + | WithSnapshotRepository(repository, repository)(d) | |
| 18 | + | d.stopped["work"] = stoppedSession{name: "work"} | |
| 19 | + | ||
| 20 | + | transport := portsmocks.NewMockTransport(t) | |
| 21 | + | transport.EXPECT().Send(mock.Anything).Return(nil).Once() | |
| 22 | + | transport.EXPECT().Close().Return(nil).Once() | |
| 23 | + | d.handleKill(transport, ports.Frame{Type: ports.MsgKill, Payload: ports.MarshalKill(ports.Kill{Name: "work"})}) | |
| 24 | + | ||
| 25 | + | require.Equal(t, []string{"tombstone", "incremental", "legacy"}, repository.calls) | |
| 26 | + | require.True(t, repository.tombstoned["work"]) | |
| 27 | + | require.True(t, d.stopped["work"].purging) | |
| 28 | + | } | |
| 29 | + | ||
| 30 | + | func TestStoppedPurgeMetadataFailureRemainsFencedForRetry(t *testing.T) { | |
| 31 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | |
| 32 | + | repository := &retryablePurgeRepository{} | |
| 33 | + | WithSnapshotRepository(repository, repository)(d) | |
| 34 | + | store, state := newMockStore(t) | |
| 35 | + | WithStore(store)(d) | |
| 36 | + | metadataErr := errors.New("metadata delete failed") | |
| 37 | + | state.mu.Lock() | |
| 38 | + | state.deleteErr = func(string) error { return metadataErr } | |
| 39 | + | state.mu.Unlock() | |
| 40 | + | d.stopped["work"] = stoppedSession{name: "work"} | |
| 41 | + | ||
| 42 | + | require.ErrorIs(t, d.retryStoppedPurge("work"), metadataErr) | |
| 43 | + | require.True(t, repository.tombstoned["work"]) | |
| 44 | + | require.Equal(t, []string{"tombstone", "incremental", "legacy"}, repository.calls) | |
| 45 | + | ||
| 46 | + | state.mu.Lock() | |
| 47 | + | state.deleteErr = nil | |
| 48 | + | state.mu.Unlock() | |
| 49 | + | require.NoError(t, d.retryStoppedPurge("work")) | |
| 50 | + | require.Equal(t, []string{"tombstone", "incremental", "legacy", "tombstone", "incremental", "legacy", "clear tombstone"}, repository.calls) | |
| 51 | + | } | |
| @@ -0,0 +1,51 @@ | |||||
| 1 | + | package daemon | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "errors" | |||
| 5 | + | "testing" | |||
| 6 | + | ||||
| 7 | + | "github.com/stretchr/testify/mock" | |||
| 8 | + | "github.com/stretchr/testify/require" | |||
| 9 | + | ||||
| 10 | + | "github.com/bnema/vev/internal/ports" | |||
| 11 | + | portsmocks "github.com/bnema/vev/internal/ports/mocks" | |||
| 12 | + | ) | |||
| 13 | + | ||||
| 14 | + | func TestHandleKillStoppedUsesDurableBothSourcePurge(t *testing.T) { | |||
| 15 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | |||
| 16 | + | repository := &retryablePurgeRepository{legacyErr: errors.New("legacy delete failed")} | |||
| 17 | + | WithSnapshotRepository(repository, repository)(d) | |||
| 18 | + | d.stopped["work"] = stoppedSession{name: "work"} | |||
| 19 | + | ||||
| 20 | + | transport := portsmocks.NewMockTransport(t) | |||
| 21 | + | transport.EXPECT().Send(mock.Anything).Return(nil).Once() | |||
| 22 | + | transport.EXPECT().Close().Return(nil).Once() | |||
| 23 | + | d.handleKill(transport, ports.Frame{Type: ports.MsgKill, Payload: ports.MarshalKill(ports.Kill{Name: "work"})}) | |||
| 24 | + | ||||
| 25 | + | require.Equal(t, []string{"tombstone", "incremental", "legacy"}, repository.calls) | |||
| 26 | + | require.True(t, repository.tombstoned["work"]) | |||
| 27 | + | require.True(t, d.stopped["work"].purging) | |||
| 28 | + | } | |||
| 29 | + | ||||
| 30 | + | func TestStoppedPurgeMetadataFailureRemainsFencedForRetry(t *testing.T) { | |||
| 31 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | |||
| 32 | + | repository := &retryablePurgeRepository{} | |||
| 33 | + | WithSnapshotRepository(repository, repository)(d) | |||
| 34 | + | store, state := newMockStore(t) | |||
| 35 | + | WithStore(store)(d) | |||
| 36 | + | metadataErr := errors.New("metadata delete failed") | |||
| 37 | + | state.mu.Lock() | |||
| 38 | + | state.deleteErr = func(string) error { return metadataErr } | |||
| 39 | + | state.mu.Unlock() | |||
| 40 | + | d.stopped["work"] = stoppedSession{name: "work"} | |||
| 41 | + | ||||
| 42 | + | require.ErrorIs(t, d.retryStoppedPurge("work"), metadataErr) | |||
| 43 | + | require.True(t, repository.tombstoned["work"]) | |||
| 44 | + | require.Equal(t, []string{"tombstone", "incremental", "legacy"}, repository.calls) | |||
| 45 | + | ||||
| 46 | + | state.mu.Lock() | |||
| 47 | + | state.deleteErr = nil | |||
| 48 | + | state.mu.Unlock() | |||
| 49 | + | require.NoError(t, d.retryStoppedPurge("work")) | |||
| 50 | + | require.Equal(t, []string{"tombstone", "incremental", "legacy", "tombstone", "incremental", "legacy", "clear tombstone"}, repository.calls) | |||
| 51 | + | } | |||
A internal/usecase/daemon/session_rename_purge_test.go +45 -0
| @@ -0,0 +1,45 @@ | |||
| 1 | + | package daemon | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "errors" | |
| 5 | + | "testing" | |
| 6 | + | ||
| 7 | + | "github.com/stretchr/testify/require" | |
| 8 | + | ||
| 9 | + | "github.com/bnema/vev/internal/domain" | |
| 10 | + | portsmocks "github.com/bnema/vev/internal/ports/mocks" | |
| 11 | + | ) | |
| 12 | + | ||
| 13 | + | func TestRenamePurgesOldIncrementalAndLegacyBeforeCommittingNewIdentity(t *testing.T) { | |
| 14 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | |
| 15 | + | repository := &retryablePurgeRepository{} | |
| 16 | + | WithSnapshotRepository(repository, repository)(d) | |
| 17 | + | store, state := newMockStore(t) | |
| 18 | + | WithStore(store)(d) | |
| 19 | + | sess := newSnapshotTestSession(t, "old", false, "/work") | |
| 20 | + | d.sessions = map[domain.SessionID]*session{sess.id: sess} | |
| 21 | + | require.NoError(t, d.persist.Save(sess.persistRecordLocked(1))) | |
| 22 | + | ||
| 23 | + | require.NoError(t, d.renameSession(sess, "new")) | |
| 24 | + | require.Equal(t, []string{"tombstone", "incremental", "legacy", "clear tombstone"}, repository.calls) | |
| 25 | + | require.False(t, state.has("old")) | |
| 26 | + | require.True(t, state.has("new")) | |
| 27 | + | require.False(t, repository.tombstoned["old"]) | |
| 28 | + | } | |
| 29 | + | ||
| 30 | + | func TestRenameLegacyFailureLeavesOldIdentityFenced(t *testing.T) { | |
| 31 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | |
| 32 | + | repository := &retryablePurgeRepository{legacyErr: errors.New("legacy delete failed")} | |
| 33 | + | WithSnapshotRepository(repository, repository)(d) | |
| 34 | + | store, state := newMockStore(t) | |
| 35 | + | WithStore(store)(d) | |
| 36 | + | sess := newSnapshotTestSession(t, "old", false, "/work") | |
| 37 | + | d.sessions = map[domain.SessionID]*session{sess.id: sess} | |
| 38 | + | require.NoError(t, d.persist.Save(sess.persistRecordLocked(1))) | |
| 39 | + | ||
| 40 | + | require.Error(t, d.renameSession(sess, "new")) | |
| 41 | + | require.Equal(t, []string{"tombstone", "incremental", "legacy"}, repository.calls) | |
| 42 | + | require.True(t, repository.tombstoned["old"]) | |
| 43 | + | require.True(t, state.has("old")) | |
| 44 | + | require.False(t, state.has("new")) | |
| 45 | + | } | |
| @@ -0,0 +1,45 @@ | |||||
| 1 | + | package daemon | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "errors" | |||
| 5 | + | "testing" | |||
| 6 | + | ||||
| 7 | + | "github.com/stretchr/testify/require" | |||
| 8 | + | ||||
| 9 | + | "github.com/bnema/vev/internal/domain" | |||
| 10 | + | portsmocks "github.com/bnema/vev/internal/ports/mocks" | |||
| 11 | + | ) | |||
| 12 | + | ||||
| 13 | + | func TestRenamePurgesOldIncrementalAndLegacyBeforeCommittingNewIdentity(t *testing.T) { | |||
| 14 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | |||
| 15 | + | repository := &retryablePurgeRepository{} | |||
| 16 | + | WithSnapshotRepository(repository, repository)(d) | |||
| 17 | + | store, state := newMockStore(t) | |||
| 18 | + | WithStore(store)(d) | |||
| 19 | + | sess := newSnapshotTestSession(t, "old", false, "/work") | |||
| 20 | + | d.sessions = map[domain.SessionID]*session{sess.id: sess} | |||
| 21 | + | require.NoError(t, d.persist.Save(sess.persistRecordLocked(1))) | |||
| 22 | + | ||||
| 23 | + | require.NoError(t, d.renameSession(sess, "new")) | |||
| 24 | + | require.Equal(t, []string{"tombstone", "incremental", "legacy", "clear tombstone"}, repository.calls) | |||
| 25 | + | require.False(t, state.has("old")) | |||
| 26 | + | require.True(t, state.has("new")) | |||
| 27 | + | require.False(t, repository.tombstoned["old"]) | |||
| 28 | + | } | |||
| 29 | + | ||||
| 30 | + | func TestRenameLegacyFailureLeavesOldIdentityFenced(t *testing.T) { | |||
| 31 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | |||
| 32 | + | repository := &retryablePurgeRepository{legacyErr: errors.New("legacy delete failed")} | |||
| 33 | + | WithSnapshotRepository(repository, repository)(d) | |||
| 34 | + | store, state := newMockStore(t) | |||
| 35 | + | WithStore(store)(d) | |||
| 36 | + | sess := newSnapshotTestSession(t, "old", false, "/work") | |||
| 37 | + | d.sessions = map[domain.SessionID]*session{sess.id: sess} | |||
| 38 | + | require.NoError(t, d.persist.Save(sess.persistRecordLocked(1))) | |||
| 39 | + | ||||
| 40 | + | require.Error(t, d.renameSession(sess, "new")) | |||
| 41 | + | require.Equal(t, []string{"tombstone", "incremental", "legacy"}, repository.calls) | |||
| 42 | + | require.True(t, repository.tombstoned["old"]) | |||
| 43 | + | require.True(t, state.has("old")) | |||
| 44 | + | require.False(t, state.has("new")) | |||
| 45 | + | } | |||
M internal/usecase/daemon/snapshot_restore_test.go +1 -1
| @@ -120,7 +120,7 @@ func restoreAcceptanceSession(t *testing.T, name string) snapcodec.Session { | |||
| 120 | 120 | screen.Write([]byte("primary-visible")) | |
| 121 | 121 | screen.Write([]byte("\x1b[?1049h")) | |
| 122 | 122 | screen.Write([]byte("alternate-active")) | |
| 123 | - | visible, err := screen.MarshalPrimaryVisible() | |
| 123 | + | visible, err := screen.PrimaryVisibleSnapshot().Marshal() | |
| 124 | 124 | require.NoError(t, err) | |
| 125 | 125 | ||
| 126 | 126 | paneID := layout.PaneID("pane-7") | |
| @@ -120,7 +120,7 @@ func restoreAcceptanceSession(t *testing.T, name string) snapcodec.Session { | |||||
| 120 | screen.Write([]byte("primary-visible")) | 120 | screen.Write([]byte("primary-visible")) | ||
| 121 | screen.Write([]byte("\x1b[?1049h")) | 121 | screen.Write([]byte("\x1b[?1049h")) | ||
| 122 | screen.Write([]byte("alternate-active")) | 122 | screen.Write([]byte("alternate-active")) | ||
| 123 | - | visible, err := screen.MarshalPrimaryVisible() | 123 | + | visible, err := screen.PrimaryVisibleSnapshot().Marshal() |
| 124 | require.NoError(t, err) | 124 | require.NoError(t, err) | ||
| 125 | 125 | ||||
| 126 | paneID := layout.PaneID("pane-7") | 126 | paneID := layout.PaneID("pane-7") | ||
M internal/usecase/daemon/snapshot_scheduler_test.go +1 -3
| @@ -191,9 +191,7 @@ func TestRenameRollbackAfterOldDeleteLeavesOldCoordinatorStopped(t *testing.T) { | |||
| 191 | 191 | case d.snapshotWake <- struct{}{}: | |
| 192 | 192 | default: | |
| 193 | 193 | } | |
| 194 | - | select { | |
| 195 | - | case <-time.After(50 * time.Millisecond): | |
| 196 | - | } | |
| 194 | + | time.Sleep(50 * time.Millisecond) | |
| 197 | 195 | } | |
| 198 | 196 | ||
| 199 | 197 | func TestForcedSnapshotSchedulesSuccessorAfterRoutineInFlight(t *testing.T) { | |
| @@ -191,9 +191,7 @@ func TestRenameRollbackAfterOldDeleteLeavesOldCoordinatorStopped(t *testing.T) { | |||||
| 191 | case d.snapshotWake <- struct{}{}: | 191 | case d.snapshotWake <- struct{}{}: | ||
| 192 | default: | 192 | default: | ||
| 193 | } | 193 | } | ||
| 194 | - | select { | 194 | + | time.Sleep(50 * time.Millisecond) |
| 195 | - | case <-time.After(50 * time.Millisecond): | |||
| 196 | - | } | |||
| 197 | } | 195 | } | ||
| 198 | 196 | ||||
| 199 | func TestForcedSnapshotSchedulesSuccessorAfterRoutineInFlight(t *testing.T) { | 197 | func TestForcedSnapshotSchedulesSuccessorAfterRoutineInFlight(t *testing.T) { | ||