refactor(recovery): guard incarnation replacement
- commit
- e362c809596efd81e5c2c52bfc4ae4f3f6707873
- author
- brice · · 2026-07-27 13:12 UTC
- parent
- 816fe32
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
3 files · +259 -18
M internal/usecase/recovery/discard.go +2 -18
| @@ -4,8 +4,6 @@ import ( | |||
| 4 | 4 | "context" | |
| 5 | 5 | "errors" | |
| 6 | 6 | "fmt" | |
| 7 | - | ||
| 8 | - | "github.com/bnema/vev/internal/domain" | |
| 9 | 7 | ) | |
| 10 | 8 | ||
| 11 | 9 | var ( | |
| @@ -43,20 +41,6 @@ func (c *Coordinator) Discard(ctx context.Context, name string) error { | |||
| 43 | 41 | return ErrSessionNotBroken | |
| 44 | 42 | } | |
| 45 | 43 | ||
| 46 | - | old := record.IncarnationID | |
| 47 | - | next, err := domain.NewIncarnationID(c.random) | |
| 48 | - | if err != nil { | |
| 49 | - | return fmt.Errorf("recovery: generate replacement incarnation: %w", err) | |
| 50 | - | } | |
| 51 | - | fresh := record | |
| 52 | - | fresh.IncarnationID = next | |
| 53 | - | fresh.Committed = nil | |
| 54 | - | fresh.DegradedReason = "" | |
| 55 | - | if err := fresh.Validate(); err != nil { | |
| 56 | - | return fmt.Errorf("recovery: invalid discard replacement: %w", err) | |
| 57 | - | } | |
| 58 | - | if err := c.catalogue.Replace(name, fresh); err != nil { | |
| 59 | - | return err | |
| 60 | - | } | |
| 61 | - | return c.repository.DeleteIncarnation(ctx, old) | |
| 44 | + | _, _, err = c.replaceIncarnationLocked(ctx, record, false, "discard") | |
| 45 | + | return err | |
| 62 | 46 | } | |
| @@ -4,8 +4,6 @@ import ( | |||||
| 4 | "context" | 4 | "context" | ||
| 5 | "errors" | 5 | "errors" | ||
| 6 | "fmt" | 6 | "fmt" | ||
| 7 | - | ||||
| 8 | - | "github.com/bnema/vev/internal/domain" | |||
| 9 | ) | 7 | ) | ||
| 10 | 8 | ||||
| 11 | var ( | 9 | var ( | ||
| @@ -43,20 +41,6 @@ func (c *Coordinator) Discard(ctx context.Context, name string) error { | |||||
| 43 | return ErrSessionNotBroken | 41 | return ErrSessionNotBroken | ||
| 44 | } | 42 | } | ||
| 45 | 43 | ||||
| 46 | - | old := record.IncarnationID | 44 | + | _, _, err = c.replaceIncarnationLocked(ctx, record, false, "discard") |
| 47 | - | next, err := domain.NewIncarnationID(c.random) | 45 | + | return err |
| 48 | - | if err != nil { | |||
| 49 | - | return fmt.Errorf("recovery: generate replacement incarnation: %w", err) | |||
| 50 | - | } | |||
| 51 | - | fresh := record | |||
| 52 | - | fresh.IncarnationID = next | |||
| 53 | - | fresh.Committed = nil | |||
| 54 | - | fresh.DegradedReason = "" | |||
| 55 | - | if err := fresh.Validate(); err != nil { | |||
| 56 | - | return fmt.Errorf("recovery: invalid discard replacement: %w", err) | |||
| 57 | - | } | |||
| 58 | - | if err := c.catalogue.Replace(name, fresh); err != nil { | |||
| 59 | - | return err | |||
| 60 | - | } | |||
| 61 | - | return c.repository.DeleteIncarnation(ctx, old) | |||
| 62 | } | 46 | } | ||
A internal/usecase/recovery/reset.go +62 -0
| @@ -0,0 +1,62 @@ | |||
| 1 | + | package recovery | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | "errors" | |
| 6 | + | "fmt" | |
| 7 | + | ||
| 8 | + | "github.com/bnema/vev/internal/domain" | |
| 9 | + | ) | |
| 10 | + | ||
| 11 | + | // ResetIncompatible replaces an exact healthy checkpoint authority with a | |
| 12 | + | // fresh incarnation. Stale observations are harmless no-ops. | |
| 13 | + | func (c *Coordinator) ResetIncompatible(ctx context.Context, name string, expectedIncarnation domain.IncarnationID, expectedCheckpoint domain.CheckpointRef) (domain.CatalogueRecord, bool, error) { | |
| 14 | + | if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil || c.random == nil { | |
| 15 | + | return domain.CatalogueRecord{}, false, errors.New("recovery: incomplete incompatible-reset dependencies") | |
| 16 | + | } | |
| 17 | + | if err := ctx.Err(); err != nil { | |
| 18 | + | return domain.CatalogueRecord{}, false, err | |
| 19 | + | } | |
| 20 | + | c.mutationMu.Lock() | |
| 21 | + | defer c.mutationMu.Unlock() | |
| 22 | + | unlock := c.locks.Lock([]string{name}) | |
| 23 | + | defer unlock() | |
| 24 | + | ||
| 25 | + | record, ok, err := c.catalogue.Record(name) | |
| 26 | + | if err != nil { | |
| 27 | + | return domain.CatalogueRecord{}, false, err | |
| 28 | + | } | |
| 29 | + | if !ok || record.IncarnationID != expectedIncarnation || record.DegradedReason != "" || | |
| 30 | + | record.Committed == nil || *record.Committed != expectedCheckpoint { | |
| 31 | + | return domain.CatalogueRecord{}, false, nil | |
| 32 | + | } | |
| 33 | + | return c.replaceIncarnationLocked(ctx, record, true, "reset") | |
| 34 | + | } | |
| 35 | + | ||
| 36 | + | // replaceIncarnationLocked commits a fresh catalogue authority before removing | |
| 37 | + | // the old repository incarnation. The caller must hold mutationMu and the | |
| 38 | + | // record name's KeyLock. | |
| 39 | + | func (c *Coordinator) replaceIncarnationLocked(ctx context.Context, record domain.CatalogueRecord, clearTabNames bool, operation string) (domain.CatalogueRecord, bool, error) { | |
| 40 | + | old := record.IncarnationID | |
| 41 | + | next, err := domain.NewIncarnationID(c.random) | |
| 42 | + | if err != nil { | |
| 43 | + | return domain.CatalogueRecord{}, false, fmt.Errorf("recovery: generate replacement incarnation: %w", err) | |
| 44 | + | } | |
| 45 | + | fresh := record | |
| 46 | + | fresh.IncarnationID = next | |
| 47 | + | fresh.Committed = nil | |
| 48 | + | fresh.DegradedReason = "" | |
| 49 | + | if clearTabNames { | |
| 50 | + | fresh.TabNames = nil | |
| 51 | + | } | |
| 52 | + | if err := fresh.Validate(); err != nil { | |
| 53 | + | return domain.CatalogueRecord{}, false, fmt.Errorf("recovery: invalid %s replacement: %w", operation, err) | |
| 54 | + | } | |
| 55 | + | if err := c.catalogue.Replace(record.Name, fresh); err != nil { | |
| 56 | + | return domain.CatalogueRecord{}, false, err | |
| 57 | + | } | |
| 58 | + | if err := c.repository.DeleteIncarnation(ctx, old); err != nil { | |
| 59 | + | return fresh, true, err | |
| 60 | + | } | |
| 61 | + | return fresh, true, nil | |
| 62 | + | } | |
| @@ -0,0 +1,62 @@ | |||||
| 1 | + | package recovery | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "context" | |||
| 5 | + | "errors" | |||
| 6 | + | "fmt" | |||
| 7 | + | ||||
| 8 | + | "github.com/bnema/vev/internal/domain" | |||
| 9 | + | ) | |||
| 10 | + | ||||
| 11 | + | // ResetIncompatible replaces an exact healthy checkpoint authority with a | |||
| 12 | + | // fresh incarnation. Stale observations are harmless no-ops. | |||
| 13 | + | func (c *Coordinator) ResetIncompatible(ctx context.Context, name string, expectedIncarnation domain.IncarnationID, expectedCheckpoint domain.CheckpointRef) (domain.CatalogueRecord, bool, error) { | |||
| 14 | + | if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil || c.random == nil { | |||
| 15 | + | return domain.CatalogueRecord{}, false, errors.New("recovery: incomplete incompatible-reset dependencies") | |||
| 16 | + | } | |||
| 17 | + | if err := ctx.Err(); err != nil { | |||
| 18 | + | return domain.CatalogueRecord{}, false, err | |||
| 19 | + | } | |||
| 20 | + | c.mutationMu.Lock() | |||
| 21 | + | defer c.mutationMu.Unlock() | |||
| 22 | + | unlock := c.locks.Lock([]string{name}) | |||
| 23 | + | defer unlock() | |||
| 24 | + | ||||
| 25 | + | record, ok, err := c.catalogue.Record(name) | |||
| 26 | + | if err != nil { | |||
| 27 | + | return domain.CatalogueRecord{}, false, err | |||
| 28 | + | } | |||
| 29 | + | if !ok || record.IncarnationID != expectedIncarnation || record.DegradedReason != "" || | |||
| 30 | + | record.Committed == nil || *record.Committed != expectedCheckpoint { | |||
| 31 | + | return domain.CatalogueRecord{}, false, nil | |||
| 32 | + | } | |||
| 33 | + | return c.replaceIncarnationLocked(ctx, record, true, "reset") | |||
| 34 | + | } | |||
| 35 | + | ||||
| 36 | + | // replaceIncarnationLocked commits a fresh catalogue authority before removing | |||
| 37 | + | // the old repository incarnation. The caller must hold mutationMu and the | |||
| 38 | + | // record name's KeyLock. | |||
| 39 | + | func (c *Coordinator) replaceIncarnationLocked(ctx context.Context, record domain.CatalogueRecord, clearTabNames bool, operation string) (domain.CatalogueRecord, bool, error) { | |||
| 40 | + | old := record.IncarnationID | |||
| 41 | + | next, err := domain.NewIncarnationID(c.random) | |||
| 42 | + | if err != nil { | |||
| 43 | + | return domain.CatalogueRecord{}, false, fmt.Errorf("recovery: generate replacement incarnation: %w", err) | |||
| 44 | + | } | |||
| 45 | + | fresh := record | |||
| 46 | + | fresh.IncarnationID = next | |||
| 47 | + | fresh.Committed = nil | |||
| 48 | + | fresh.DegradedReason = "" | |||
| 49 | + | if clearTabNames { | |||
| 50 | + | fresh.TabNames = nil | |||
| 51 | + | } | |||
| 52 | + | if err := fresh.Validate(); err != nil { | |||
| 53 | + | return domain.CatalogueRecord{}, false, fmt.Errorf("recovery: invalid %s replacement: %w", operation, err) | |||
| 54 | + | } | |||
| 55 | + | if err := c.catalogue.Replace(record.Name, fresh); err != nil { | |||
| 56 | + | return domain.CatalogueRecord{}, false, err | |||
| 57 | + | } | |||
| 58 | + | if err := c.repository.DeleteIncarnation(ctx, old); err != nil { | |||
| 59 | + | return fresh, true, err | |||
| 60 | + | } | |||
| 61 | + | return fresh, true, nil | |||
| 62 | + | } | |||
M internal/usecase/recovery/transactions_test.go +195 -0
| @@ -107,6 +107,7 @@ type transactionRepository struct { | |||
| 107 | 107 | ports.SnapshotRepository | |
| 108 | 108 | mu sync.Mutex | |
| 109 | 109 | deleteErrOnce error | |
| 110 | + | deleteCalls []domain.IncarnationID | |
| 110 | 111 | deletedIDs []domain.IncarnationID | |
| 111 | 112 | deleteEntered chan struct{} | |
| 112 | 113 | releaseDelete chan struct{} | |
| @@ -204,6 +205,7 @@ func (r *transactionRepository) DeleteIncarnation(_ context.Context, id domain.I | |||
| 204 | 205 | } | |
| 205 | 206 | r.mu.Lock() | |
| 206 | 207 | defer r.mu.Unlock() | |
| 208 | + | r.deleteCalls = append(r.deleteCalls, id) | |
| 207 | 209 | if r.deleteErrOnce != nil { | |
| 208 | 210 | err := r.deleteErrOnce | |
| 209 | 211 | r.deleteErrOnce = nil | |
| @@ -220,6 +222,137 @@ func degradedTransactionRecord() domain.CatalogueRecord { | |||
| 220 | 222 | } | |
| 221 | 223 | } | |
| 222 | 224 | ||
| 225 | + | func healthyTransactionRecord() domain.CatalogueRecord { | |
| 226 | + | return domain.CatalogueRecord{ | |
| 227 | + | Name: "work", | |
| 228 | + | IncarnationID: domain.IncarnationID{1}, | |
| 229 | + | Cwd: "/workspace", | |
| 230 | + | CreatedAt: 11, | |
| 231 | + | UpdatedAt: 22, | |
| 232 | + | LastUsedSeq: 33, | |
| 233 | + | TabNames: []string{"shell", "logs"}, | |
| 234 | + | Committed: &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}}, | |
| 235 | + | } | |
| 236 | + | } | |
| 237 | + | ||
| 238 | + | func TestResetIncompatibleReplacesExactHealthyCheckpoint(t *testing.T) { | |
| 239 | + | t.Parallel() | |
| 240 | + | old := healthyTransactionRecord() | |
| 241 | + | catalogue := newTransactionCatalogue(old) | |
| 242 | + | repository := &transactionRepository{} | |
| 243 | + | coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16))) | |
| 244 | + | ||
| 245 | + | fresh, committed, err := coordinator.ResetIncompatible(context.Background(), old.Name, old.IncarnationID, *old.Committed) | |
| 246 | + | ||
| 247 | + | require.NoError(t, err) | |
| 248 | + | require.True(t, committed) | |
| 249 | + | require.NotZero(t, fresh.IncarnationID) | |
| 250 | + | require.NotEqual(t, old.IncarnationID, fresh.IncarnationID) | |
| 251 | + | require.Equal(t, old.Name, fresh.Name) | |
| 252 | + | require.Equal(t, old.Cwd, fresh.Cwd) | |
| 253 | + | require.Equal(t, old.CreatedAt, fresh.CreatedAt) | |
| 254 | + | require.Equal(t, old.UpdatedAt, fresh.UpdatedAt) | |
| 255 | + | require.Equal(t, old.LastUsedSeq, fresh.LastUsedSeq) | |
| 256 | + | require.Empty(t, fresh.TabNames) | |
| 257 | + | require.Nil(t, fresh.Committed) | |
| 258 | + | require.Empty(t, fresh.DegradedReason) | |
| 259 | + | stored, ok, recordErr := catalogue.Record(old.Name) | |
| 260 | + | require.NoError(t, recordErr) | |
| 261 | + | require.True(t, ok) | |
| 262 | + | require.True(t, fresh.Equal(stored)) | |
| 263 | + | require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deleteCalls) | |
| 264 | + | require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deletedIDs) | |
| 265 | + | } | |
| 266 | + | ||
| 267 | + | func TestResetIncompatibleDeleteFailureReturnsCommittedFreshAuthority(t *testing.T) { | |
| 268 | + | t.Parallel() | |
| 269 | + | old := healthyTransactionRecord() | |
| 270 | + | cause := errors.New("repository delete failed") | |
| 271 | + | catalogue := newTransactionCatalogue(old) | |
| 272 | + | repository := &transactionRepository{deleteErrOnce: cause} | |
| 273 | + | coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16))) | |
| 274 | + | ||
| 275 | + | fresh, committed, err := coordinator.ResetIncompatible(context.Background(), old.Name, old.IncarnationID, *old.Committed) | |
| 276 | + | ||
| 277 | + | require.ErrorIs(t, err, cause) | |
| 278 | + | require.True(t, committed) | |
| 279 | + | require.NotEqual(t, old.IncarnationID, fresh.IncarnationID) | |
| 280 | + | stored, ok, recordErr := catalogue.Record(old.Name) | |
| 281 | + | require.NoError(t, recordErr) | |
| 282 | + | require.True(t, ok) | |
| 283 | + | require.True(t, fresh.Equal(stored)) | |
| 284 | + | require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deleteCalls) | |
| 285 | + | require.Empty(t, repository.deletedIDs) | |
| 286 | + | } | |
| 287 | + | ||
| 288 | + | func TestResetIncompatibleReplaceFailurePreservesOldAuthorityAndRepository(t *testing.T) { | |
| 289 | + | t.Parallel() | |
| 290 | + | old := healthyTransactionRecord() | |
| 291 | + | cause := errors.New("catalogue replace failed") | |
| 292 | + | catalogue := newTransactionCatalogue(old) | |
| 293 | + | catalogue.replaceErrOnce = cause | |
| 294 | + | repository := &transactionRepository{} | |
| 295 | + | coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16))) | |
| 296 | + | ||
| 297 | + | fresh, committed, err := coordinator.ResetIncompatible(context.Background(), old.Name, old.IncarnationID, *old.Committed) | |
| 298 | + | ||
| 299 | + | require.ErrorIs(t, err, cause) | |
| 300 | + | require.False(t, committed) | |
| 301 | + | require.Equal(t, domain.CatalogueRecord{}, fresh) | |
| 302 | + | stored, ok, recordErr := catalogue.Record(old.Name) | |
| 303 | + | require.NoError(t, recordErr) | |
| 304 | + | require.True(t, ok) | |
| 305 | + | require.True(t, old.Equal(stored)) | |
| 306 | + | require.Empty(t, repository.deleteCalls) | |
| 307 | + | } | |
| 308 | + | ||
| 309 | + | func TestResetIncompatibleStaleAuthorityIsNoOp(t *testing.T) { | |
| 310 | + | t.Parallel() | |
| 311 | + | old := healthyTransactionRecord() | |
| 312 | + | otherGeneration := *old.Committed | |
| 313 | + | otherGeneration.Generation++ | |
| 314 | + | otherDigest := *old.Committed | |
| 315 | + | otherDigest.ManifestDigest[0]++ | |
| 316 | + | uncommitted := old | |
| 317 | + | uncommitted.Committed = nil | |
| 318 | + | tests := []struct { | |
| 319 | + | name string | |
| 320 | + | record *domain.CatalogueRecord | |
| 321 | + | expected domain.IncarnationID | |
| 322 | + | ref domain.CheckpointRef | |
| 323 | + | }{ | |
| 324 | + | {name: "missing record", expected: old.IncarnationID, ref: *old.Committed}, | |
| 325 | + | {name: "different incarnation", record: &old, expected: domain.IncarnationID{9}, ref: *old.Committed}, | |
| 326 | + | {name: "missing checkpoint", record: &uncommitted, expected: old.IncarnationID, ref: *old.Committed}, | |
| 327 | + | {name: "different checkpoint generation", record: &old, expected: old.IncarnationID, ref: otherGeneration}, | |
| 328 | + | {name: "different checkpoint digest", record: &old, expected: old.IncarnationID, ref: otherDigest}, | |
| 329 | + | {name: "degraded record", record: func() *domain.CatalogueRecord { | |
| 330 | + | degraded := old | |
| 331 | + | degraded.DegradedReason = "restore failed" | |
| 332 | + | return °raded | |
| 333 | + | }(), expected: old.IncarnationID, ref: *old.Committed}, | |
| 334 | + | } | |
| 335 | + | for _, tt := range tests { | |
| 336 | + | t.Run(tt.name, func(t *testing.T) { | |
| 337 | + | t.Parallel() | |
| 338 | + | catalogue := newTransactionCatalogue() | |
| 339 | + | if tt.record != nil { | |
| 340 | + | catalogue.records[tt.record.Name] = *tt.record | |
| 341 | + | } | |
| 342 | + | repository := &transactionRepository{} | |
| 343 | + | coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16))) | |
| 344 | + | ||
| 345 | + | fresh, committed, err := coordinator.ResetIncompatible(context.Background(), old.Name, tt.expected, tt.ref) | |
| 346 | + | ||
| 347 | + | require.NoError(t, err) | |
| 348 | + | require.False(t, committed) | |
| 349 | + | require.Equal(t, domain.CatalogueRecord{}, fresh) | |
| 350 | + | require.Empty(t, catalogue.events) | |
| 351 | + | require.Empty(t, repository.deleteCalls) | |
| 352 | + | }) | |
| 353 | + | } | |
| 354 | + | } | |
| 355 | + | ||
| 223 | 356 | func TestStartupGarbageCollectionFencesNewIncarnationsAndCheckpointPublications(t *testing.T) { | |
| 224 | 357 | t.Parallel() | |
| 225 | 358 | ||
| @@ -338,6 +471,66 @@ func TestMarkBrokenCannotOverwriteDiscardReplacement(t *testing.T) { | |||
| 338 | 471 | require.Empty(t, got.DegradedReason) | |
| 339 | 472 | } | |
| 340 | 473 | ||
| 474 | + | func TestDiscardPublicPreconditions(t *testing.T) { | |
| 475 | + | t.Parallel() | |
| 476 | + | uncommitted := degradedTransactionRecord() | |
| 477 | + | uncommitted.Name = "work" | |
| 478 | + | uncommitted.Committed = nil | |
| 479 | + | uncommitted.DegradedReason = "" | |
| 480 | + | healthy := healthyTransactionRecord() | |
| 481 | + | tests := []struct { | |
| 482 | + | name string | |
| 483 | + | record *domain.CatalogueRecord | |
| 484 | + | incomplete bool | |
| 485 | + | cancelled bool | |
| 486 | + | wantErr error | |
| 487 | + | }{ | |
| 488 | + | {name: "incomplete dependencies", incomplete: true, wantErr: errors.New("recovery: incomplete discard dependencies")}, | |
| 489 | + | {name: "cancelled context", record: &healthy, cancelled: true, wantErr: context.Canceled}, | |
| 490 | + | {name: "missing record", wantErr: ErrRecoveryRecordNotFound}, | |
| 491 | + | {name: "already fresh incarnation", record: &uncommitted}, | |
| 492 | + | {name: "healthy checkpoint", record: &healthy, wantErr: ErrSessionNotBroken}, | |
| 493 | + | } | |
| 494 | + | for _, tt := range tests { | |
| 495 | + | t.Run(tt.name, func(t *testing.T) { | |
| 496 | + | t.Parallel() | |
| 497 | + | ctx := context.Background() | |
| 498 | + | if tt.cancelled { | |
| 499 | + | cancelled, cancel := context.WithCancel(ctx) | |
| 500 | + | cancel() | |
| 501 | + | ctx = cancelled | |
| 502 | + | } | |
| 503 | + | var coordinator *Coordinator | |
| 504 | + | var catalogue *transactionCatalogue | |
| 505 | + | var repository *transactionRepository | |
| 506 | + | if tt.incomplete { | |
| 507 | + | coordinator = &Coordinator{} | |
| 508 | + | } else { | |
| 509 | + | catalogue = newTransactionCatalogue() | |
| 510 | + | if tt.record != nil { | |
| 511 | + | catalogue.records[tt.record.Name] = *tt.record | |
| 512 | + | } | |
| 513 | + | repository = &transactionRepository{} | |
| 514 | + | coordinator = NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16))) | |
| 515 | + | } | |
| 516 | + | ||
| 517 | + | err := coordinator.Discard(ctx, "work") | |
| 518 | + | ||
| 519 | + | if tt.wantErr == nil { | |
| 520 | + | require.NoError(t, err) | |
| 521 | + | } else if tt.incomplete { | |
| 522 | + | require.EqualError(t, err, tt.wantErr.Error()) | |
| 523 | + | } else { | |
| 524 | + | require.ErrorIs(t, err, tt.wantErr) | |
| 525 | + | } | |
| 526 | + | if !tt.incomplete { | |
| 527 | + | require.Empty(t, catalogue.events) | |
| 528 | + | require.Empty(t, repository.deleteCalls) | |
| 529 | + | } | |
| 530 | + | }) | |
| 531 | + | } | |
| 532 | + | } | |
| 533 | + | ||
| 341 | 534 | func TestDiscardIsRetryIdempotent(t *testing.T) { | |
| 342 | 535 | t.Parallel() | |
| 343 | 536 | failure := errors.New("injected discard failure") | |
| @@ -352,6 +545,7 @@ func TestDiscardIsRetryIdempotent(t *testing.T) { | |||
| 352 | 545 | t.Run(tt.name, func(t *testing.T) { | |
| 353 | 546 | t.Parallel() | |
| 354 | 547 | old := degradedTransactionRecord() | |
| 548 | + | old.TabNames = []string{"shell"} | |
| 355 | 549 | catalogue := newTransactionCatalogue(old) | |
| 356 | 550 | repository := &transactionRepository{} | |
| 357 | 551 | if tt.failBefore { | |
| @@ -371,6 +565,7 @@ func TestDiscardIsRetryIdempotent(t *testing.T) { | |||
| 371 | 565 | require.True(t, ok) | |
| 372 | 566 | require.Nil(t, got.Committed) | |
| 373 | 567 | require.NotEqual(t, old.IncarnationID, got.IncarnationID) | |
| 568 | + | require.Equal(t, old.TabNames, got.TabNames) | |
| 374 | 569 | if tt.failBefore { | |
| 375 | 570 | require.Contains(t, repository.deletedIDs, old.IncarnationID) | |
| 376 | 571 | } else { | |
| @@ -107,6 +107,7 @@ type transactionRepository struct { | |||||
| 107 | ports.SnapshotRepository | 107 | ports.SnapshotRepository | ||
| 108 | mu sync.Mutex | 108 | mu sync.Mutex | ||
| 109 | deleteErrOnce error | 109 | deleteErrOnce error | ||
| 110 | + | deleteCalls []domain.IncarnationID | |||
| 110 | deletedIDs []domain.IncarnationID | 111 | deletedIDs []domain.IncarnationID | ||
| 111 | deleteEntered chan struct{} | 112 | deleteEntered chan struct{} | ||
| 112 | releaseDelete chan struct{} | 113 | releaseDelete chan struct{} | ||
| @@ -204,6 +205,7 @@ func (r *transactionRepository) DeleteIncarnation(_ context.Context, id domain.I | |||||
| 204 | } | 205 | } | ||
| 205 | r.mu.Lock() | 206 | r.mu.Lock() | ||
| 206 | defer r.mu.Unlock() | 207 | defer r.mu.Unlock() | ||
| 208 | + | r.deleteCalls = append(r.deleteCalls, id) | |||
| 207 | if r.deleteErrOnce != nil { | 209 | if r.deleteErrOnce != nil { | ||
| 208 | err := r.deleteErrOnce | 210 | err := r.deleteErrOnce | ||
| 209 | r.deleteErrOnce = nil | 211 | r.deleteErrOnce = nil | ||
| @@ -220,6 +222,137 @@ func degradedTransactionRecord() domain.CatalogueRecord { | |||||
| 220 | } | 222 | } | ||
| 221 | } | 223 | } | ||
| 222 | 224 | ||||
| 225 | + | func healthyTransactionRecord() domain.CatalogueRecord { | |||
| 226 | + | return domain.CatalogueRecord{ | |||
| 227 | + | Name: "work", | |||
| 228 | + | IncarnationID: domain.IncarnationID{1}, | |||
| 229 | + | Cwd: "/workspace", | |||
| 230 | + | CreatedAt: 11, | |||
| 231 | + | UpdatedAt: 22, | |||
| 232 | + | LastUsedSeq: 33, | |||
| 233 | + | TabNames: []string{"shell", "logs"}, | |||
| 234 | + | Committed: &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}}, | |||
| 235 | + | } | |||
| 236 | + | } | |||
| 237 | + | ||||
| 238 | + | func TestResetIncompatibleReplacesExactHealthyCheckpoint(t *testing.T) { | |||
| 239 | + | t.Parallel() | |||
| 240 | + | old := healthyTransactionRecord() | |||
| 241 | + | catalogue := newTransactionCatalogue(old) | |||
| 242 | + | repository := &transactionRepository{} | |||
| 243 | + | coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16))) | |||
| 244 | + | ||||
| 245 | + | fresh, committed, err := coordinator.ResetIncompatible(context.Background(), old.Name, old.IncarnationID, *old.Committed) | |||
| 246 | + | ||||
| 247 | + | require.NoError(t, err) | |||
| 248 | + | require.True(t, committed) | |||
| 249 | + | require.NotZero(t, fresh.IncarnationID) | |||
| 250 | + | require.NotEqual(t, old.IncarnationID, fresh.IncarnationID) | |||
| 251 | + | require.Equal(t, old.Name, fresh.Name) | |||
| 252 | + | require.Equal(t, old.Cwd, fresh.Cwd) | |||
| 253 | + | require.Equal(t, old.CreatedAt, fresh.CreatedAt) | |||
| 254 | + | require.Equal(t, old.UpdatedAt, fresh.UpdatedAt) | |||
| 255 | + | require.Equal(t, old.LastUsedSeq, fresh.LastUsedSeq) | |||
| 256 | + | require.Empty(t, fresh.TabNames) | |||
| 257 | + | require.Nil(t, fresh.Committed) | |||
| 258 | + | require.Empty(t, fresh.DegradedReason) | |||
| 259 | + | stored, ok, recordErr := catalogue.Record(old.Name) | |||
| 260 | + | require.NoError(t, recordErr) | |||
| 261 | + | require.True(t, ok) | |||
| 262 | + | require.True(t, fresh.Equal(stored)) | |||
| 263 | + | require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deleteCalls) | |||
| 264 | + | require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deletedIDs) | |||
| 265 | + | } | |||
| 266 | + | ||||
| 267 | + | func TestResetIncompatibleDeleteFailureReturnsCommittedFreshAuthority(t *testing.T) { | |||
| 268 | + | t.Parallel() | |||
| 269 | + | old := healthyTransactionRecord() | |||
| 270 | + | cause := errors.New("repository delete failed") | |||
| 271 | + | catalogue := newTransactionCatalogue(old) | |||
| 272 | + | repository := &transactionRepository{deleteErrOnce: cause} | |||
| 273 | + | coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16))) | |||
| 274 | + | ||||
| 275 | + | fresh, committed, err := coordinator.ResetIncompatible(context.Background(), old.Name, old.IncarnationID, *old.Committed) | |||
| 276 | + | ||||
| 277 | + | require.ErrorIs(t, err, cause) | |||
| 278 | + | require.True(t, committed) | |||
| 279 | + | require.NotEqual(t, old.IncarnationID, fresh.IncarnationID) | |||
| 280 | + | stored, ok, recordErr := catalogue.Record(old.Name) | |||
| 281 | + | require.NoError(t, recordErr) | |||
| 282 | + | require.True(t, ok) | |||
| 283 | + | require.True(t, fresh.Equal(stored)) | |||
| 284 | + | require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deleteCalls) | |||
| 285 | + | require.Empty(t, repository.deletedIDs) | |||
| 286 | + | } | |||
| 287 | + | ||||
| 288 | + | func TestResetIncompatibleReplaceFailurePreservesOldAuthorityAndRepository(t *testing.T) { | |||
| 289 | + | t.Parallel() | |||
| 290 | + | old := healthyTransactionRecord() | |||
| 291 | + | cause := errors.New("catalogue replace failed") | |||
| 292 | + | catalogue := newTransactionCatalogue(old) | |||
| 293 | + | catalogue.replaceErrOnce = cause | |||
| 294 | + | repository := &transactionRepository{} | |||
| 295 | + | coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16))) | |||
| 296 | + | ||||
| 297 | + | fresh, committed, err := coordinator.ResetIncompatible(context.Background(), old.Name, old.IncarnationID, *old.Committed) | |||
| 298 | + | ||||
| 299 | + | require.ErrorIs(t, err, cause) | |||
| 300 | + | require.False(t, committed) | |||
| 301 | + | require.Equal(t, domain.CatalogueRecord{}, fresh) | |||
| 302 | + | stored, ok, recordErr := catalogue.Record(old.Name) | |||
| 303 | + | require.NoError(t, recordErr) | |||
| 304 | + | require.True(t, ok) | |||
| 305 | + | require.True(t, old.Equal(stored)) | |||
| 306 | + | require.Empty(t, repository.deleteCalls) | |||
| 307 | + | } | |||
| 308 | + | ||||
| 309 | + | func TestResetIncompatibleStaleAuthorityIsNoOp(t *testing.T) { | |||
| 310 | + | t.Parallel() | |||
| 311 | + | old := healthyTransactionRecord() | |||
| 312 | + | otherGeneration := *old.Committed | |||
| 313 | + | otherGeneration.Generation++ | |||
| 314 | + | otherDigest := *old.Committed | |||
| 315 | + | otherDigest.ManifestDigest[0]++ | |||
| 316 | + | uncommitted := old | |||
| 317 | + | uncommitted.Committed = nil | |||
| 318 | + | tests := []struct { | |||
| 319 | + | name string | |||
| 320 | + | record *domain.CatalogueRecord | |||
| 321 | + | expected domain.IncarnationID | |||
| 322 | + | ref domain.CheckpointRef | |||
| 323 | + | }{ | |||
| 324 | + | {name: "missing record", expected: old.IncarnationID, ref: *old.Committed}, | |||
| 325 | + | {name: "different incarnation", record: &old, expected: domain.IncarnationID{9}, ref: *old.Committed}, | |||
| 326 | + | {name: "missing checkpoint", record: &uncommitted, expected: old.IncarnationID, ref: *old.Committed}, | |||
| 327 | + | {name: "different checkpoint generation", record: &old, expected: old.IncarnationID, ref: otherGeneration}, | |||
| 328 | + | {name: "different checkpoint digest", record: &old, expected: old.IncarnationID, ref: otherDigest}, | |||
| 329 | + | {name: "degraded record", record: func() *domain.CatalogueRecord { | |||
| 330 | + | degraded := old | |||
| 331 | + | degraded.DegradedReason = "restore failed" | |||
| 332 | + | return °raded | |||
| 333 | + | }(), expected: old.IncarnationID, ref: *old.Committed}, | |||
| 334 | + | } | |||
| 335 | + | for _, tt := range tests { | |||
| 336 | + | t.Run(tt.name, func(t *testing.T) { | |||
| 337 | + | t.Parallel() | |||
| 338 | + | catalogue := newTransactionCatalogue() | |||
| 339 | + | if tt.record != nil { | |||
| 340 | + | catalogue.records[tt.record.Name] = *tt.record | |||
| 341 | + | } | |||
| 342 | + | repository := &transactionRepository{} | |||
| 343 | + | coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16))) | |||
| 344 | + | ||||
| 345 | + | fresh, committed, err := coordinator.ResetIncompatible(context.Background(), old.Name, tt.expected, tt.ref) | |||
| 346 | + | ||||
| 347 | + | require.NoError(t, err) | |||
| 348 | + | require.False(t, committed) | |||
| 349 | + | require.Equal(t, domain.CatalogueRecord{}, fresh) | |||
| 350 | + | require.Empty(t, catalogue.events) | |||
| 351 | + | require.Empty(t, repository.deleteCalls) | |||
| 352 | + | }) | |||
| 353 | + | } | |||
| 354 | + | } | |||
| 355 | + | ||||
| 223 | func TestStartupGarbageCollectionFencesNewIncarnationsAndCheckpointPublications(t *testing.T) { | 356 | func TestStartupGarbageCollectionFencesNewIncarnationsAndCheckpointPublications(t *testing.T) { | ||
| 224 | t.Parallel() | 357 | t.Parallel() | ||
| 225 | 358 | ||||
| @@ -338,6 +471,66 @@ func TestMarkBrokenCannotOverwriteDiscardReplacement(t *testing.T) { | |||||
| 338 | require.Empty(t, got.DegradedReason) | 471 | require.Empty(t, got.DegradedReason) | ||
| 339 | } | 472 | } | ||
| 340 | 473 | ||||
| 474 | + | func TestDiscardPublicPreconditions(t *testing.T) { | |||
| 475 | + | t.Parallel() | |||
| 476 | + | uncommitted := degradedTransactionRecord() | |||
| 477 | + | uncommitted.Name = "work" | |||
| 478 | + | uncommitted.Committed = nil | |||
| 479 | + | uncommitted.DegradedReason = "" | |||
| 480 | + | healthy := healthyTransactionRecord() | |||
| 481 | + | tests := []struct { | |||
| 482 | + | name string | |||
| 483 | + | record *domain.CatalogueRecord | |||
| 484 | + | incomplete bool | |||
| 485 | + | cancelled bool | |||
| 486 | + | wantErr error | |||
| 487 | + | }{ | |||
| 488 | + | {name: "incomplete dependencies", incomplete: true, wantErr: errors.New("recovery: incomplete discard dependencies")}, | |||
| 489 | + | {name: "cancelled context", record: &healthy, cancelled: true, wantErr: context.Canceled}, | |||
| 490 | + | {name: "missing record", wantErr: ErrRecoveryRecordNotFound}, | |||
| 491 | + | {name: "already fresh incarnation", record: &uncommitted}, | |||
| 492 | + | {name: "healthy checkpoint", record: &healthy, wantErr: ErrSessionNotBroken}, | |||
| 493 | + | } | |||
| 494 | + | for _, tt := range tests { | |||
| 495 | + | t.Run(tt.name, func(t *testing.T) { | |||
| 496 | + | t.Parallel() | |||
| 497 | + | ctx := context.Background() | |||
| 498 | + | if tt.cancelled { | |||
| 499 | + | cancelled, cancel := context.WithCancel(ctx) | |||
| 500 | + | cancel() | |||
| 501 | + | ctx = cancelled | |||
| 502 | + | } | |||
| 503 | + | var coordinator *Coordinator | |||
| 504 | + | var catalogue *transactionCatalogue | |||
| 505 | + | var repository *transactionRepository | |||
| 506 | + | if tt.incomplete { | |||
| 507 | + | coordinator = &Coordinator{} | |||
| 508 | + | } else { | |||
| 509 | + | catalogue = newTransactionCatalogue() | |||
| 510 | + | if tt.record != nil { | |||
| 511 | + | catalogue.records[tt.record.Name] = *tt.record | |||
| 512 | + | } | |||
| 513 | + | repository = &transactionRepository{} | |||
| 514 | + | coordinator = NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16))) | |||
| 515 | + | } | |||
| 516 | + | ||||
| 517 | + | err := coordinator.Discard(ctx, "work") | |||
| 518 | + | ||||
| 519 | + | if tt.wantErr == nil { | |||
| 520 | + | require.NoError(t, err) | |||
| 521 | + | } else if tt.incomplete { | |||
| 522 | + | require.EqualError(t, err, tt.wantErr.Error()) | |||
| 523 | + | } else { | |||
| 524 | + | require.ErrorIs(t, err, tt.wantErr) | |||
| 525 | + | } | |||
| 526 | + | if !tt.incomplete { | |||
| 527 | + | require.Empty(t, catalogue.events) | |||
| 528 | + | require.Empty(t, repository.deleteCalls) | |||
| 529 | + | } | |||
| 530 | + | }) | |||
| 531 | + | } | |||
| 532 | + | } | |||
| 533 | + | ||||
| 341 | func TestDiscardIsRetryIdempotent(t *testing.T) { | 534 | func TestDiscardIsRetryIdempotent(t *testing.T) { | ||
| 342 | t.Parallel() | 535 | t.Parallel() | ||
| 343 | failure := errors.New("injected discard failure") | 536 | failure := errors.New("injected discard failure") | ||
| @@ -352,6 +545,7 @@ func TestDiscardIsRetryIdempotent(t *testing.T) { | |||||
| 352 | t.Run(tt.name, func(t *testing.T) { | 545 | t.Run(tt.name, func(t *testing.T) { | ||
| 353 | t.Parallel() | 546 | t.Parallel() | ||
| 354 | old := degradedTransactionRecord() | 547 | old := degradedTransactionRecord() | ||
| 548 | + | old.TabNames = []string{"shell"} | |||
| 355 | catalogue := newTransactionCatalogue(old) | 549 | catalogue := newTransactionCatalogue(old) | ||
| 356 | repository := &transactionRepository{} | 550 | repository := &transactionRepository{} | ||
| 357 | if tt.failBefore { | 551 | if tt.failBefore { | ||
| @@ -371,6 +565,7 @@ func TestDiscardIsRetryIdempotent(t *testing.T) { | |||||
| 371 | require.True(t, ok) | 565 | require.True(t, ok) | ||
| 372 | require.Nil(t, got.Committed) | 566 | require.Nil(t, got.Committed) | ||
| 373 | require.NotEqual(t, old.IncarnationID, got.IncarnationID) | 567 | require.NotEqual(t, old.IncarnationID, got.IncarnationID) | ||
| 568 | + | require.Equal(t, old.TabNames, got.TabNames) | |||
| 374 | if tt.failBefore { | 569 | if tt.failBefore { | ||
| 375 | require.Contains(t, repository.deletedIDs, old.IncarnationID) | 570 | require.Contains(t, repository.deletedIDs, old.IncarnationID) | ||
| 376 | } else { | 571 | } else { | ||