test(daemon): cover checkpoint scheduler semantics
- commit
- 6c78be9df75277b2579ac1a31a3c53c41147b34d
- author
- brice · · 2026-07-21 22:50 UTC
- parent
- d1419bc
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
1 file · +263 -0
M internal/usecase/daemon/snapshot_scheduler_test.go +263 -0
| @@ -451,3 +451,266 @@ func (t *snapshotDeadlineTimer) C() <-chan time.Time { return t.ch } | |||
| 451 | 451 | func (*snapshotDeadlineTimer) Reset(time.Duration) bool { return true } | |
| 452 | 452 | func (*snapshotDeadlineTimer) Stop() bool { return true } | |
| 453 | 453 | func (t *snapshotDeadlineTimer) fire() { t.ch <- time.Unix(101, 0) } | |
| 454 | + | ||
| 455 | + | // deterministicSnapshotClock records timer resets and advances only when a | |
| 456 | + | // test explicitly fires its timer. It deliberately has no wall-clock behavior. | |
| 457 | + | type deterministicSnapshotClock struct { | |
| 458 | + | mu sync.Mutex | |
| 459 | + | now time.Time | |
| 460 | + | timers chan *deterministicSnapshotTimer | |
| 461 | + | } | |
| 462 | + | ||
| 463 | + | func newDeterministicSnapshotClock(now time.Time) *deterministicSnapshotClock { | |
| 464 | + | return &deterministicSnapshotClock{now: now, timers: make(chan *deterministicSnapshotTimer, 1)} | |
| 465 | + | } | |
| 466 | + | ||
| 467 | + | func (c *deterministicSnapshotClock) Now() time.Time { | |
| 468 | + | c.mu.Lock() | |
| 469 | + | defer c.mu.Unlock() | |
| 470 | + | return c.now | |
| 471 | + | } | |
| 472 | + | ||
| 473 | + | func (c *deterministicSnapshotClock) advance(now time.Time) { | |
| 474 | + | c.mu.Lock() | |
| 475 | + | c.now = now | |
| 476 | + | c.mu.Unlock() | |
| 477 | + | } | |
| 478 | + | ||
| 479 | + | func (c *deterministicSnapshotClock) NewTimer(time.Duration) ports.Timer { | |
| 480 | + | timer := &deterministicSnapshotTimer{clock: c, ch: make(chan time.Time, 1), resets: make(chan time.Duration, 16)} | |
| 481 | + | c.timers <- timer | |
| 482 | + | return timer | |
| 483 | + | } | |
| 484 | + | ||
| 485 | + | func (c *deterministicSnapshotClock) timer() *deterministicSnapshotTimer { return <-c.timers } | |
| 486 | + | ||
| 487 | + | type deterministicSnapshotTimer struct { | |
| 488 | + | clock *deterministicSnapshotClock | |
| 489 | + | ch chan time.Time | |
| 490 | + | resets chan time.Duration | |
| 491 | + | } | |
| 492 | + | ||
| 493 | + | func (t *deterministicSnapshotTimer) C() <-chan time.Time { return t.ch } | |
| 494 | + | func (t *deterministicSnapshotTimer) Reset(delay time.Duration) bool { | |
| 495 | + | t.resets <- delay | |
| 496 | + | return true | |
| 497 | + | } | |
| 498 | + | func (*deterministicSnapshotTimer) Stop() bool { return true } | |
| 499 | + | func (t *deterministicSnapshotTimer) fire() { t.ch <- t.clock.Now() } | |
| 500 | + | ||
| 501 | + | func awaitSnapshotCleanSignal(sess *session) { | |
| 502 | + | for { | |
| 503 | + | sess.snapshotMu.Lock() | |
| 504 | + | clean := !sess.snapDirty.Load() | |
| 505 | + | changed := sess.snapshotChangeLocked() | |
| 506 | + | sess.snapshotMu.Unlock() | |
| 507 | + | if clean { | |
| 508 | + | return | |
| 509 | + | } | |
| 510 | + | <-changed | |
| 511 | + | } | |
| 512 | + | } | |
| 513 | + | ||
| 514 | + | func TestSnapshotCompletionSchedulesRoutineRetriesFromCompletion(t *testing.T) { | |
| 515 | + | now := time.Unix(100, 0) | |
| 516 | + | clock := newDeterministicSnapshotClock(now) | |
| 517 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock) | |
| 518 | + | repository := portsmocks.NewMockSnapshotRepository(t) | |
| 519 | + | WithSnapshotRepository(repository, nil)(d) | |
| 520 | + | ||
| 521 | + | for _, tt := range []struct { | |
| 522 | + | name string | |
| 523 | + | kind snapshotAttemptKind | |
| 524 | + | succeeded bool | |
| 525 | + | existingEligible time.Time | |
| 526 | + | wantDirty bool | |
| 527 | + | wantEligible time.Time | |
| 528 | + | }{ | |
| 529 | + | {name: "routine success", kind: snapshotAttemptRoutine, succeeded: true, wantEligible: now.Add(snapshotInterval)}, | |
| 530 | + | {name: "routine failure", kind: snapshotAttemptRoutine, wantDirty: true, wantEligible: now.Add(snapshotInterval)}, | |
| 531 | + | {name: "forced completion preserves routine eligibility", kind: snapshotAttemptForced, succeeded: true, existingEligible: now.Add(time.Minute), wantEligible: now.Add(time.Minute)}, | |
| 532 | + | } { | |
| 533 | + | t.Run(tt.name, func(t *testing.T) { | |
| 534 | + | sess := newSnapshotTestSession(t, "work", false, "/work") | |
| 535 | + | sess.snapshotGeneration = 1 | |
| 536 | + | sess.snapshotNextEligibleAt = tt.existingEligible | |
| 537 | + | sess.snapDirty.Store(true) | |
| 538 | + | d.finishSnapshotCapture(&snapshotCapture{session: sess, generation: 1, attemptKind: tt.kind}, tt.succeeded) | |
| 539 | + | ||
| 540 | + | require.Equal(t, tt.wantDirty, sess.snapDirty.Load()) | |
| 541 | + | sess.snapshotMu.Lock() | |
| 542 | + | require.Equal(t, tt.wantEligible, sess.snapshotNextEligibleAt) | |
| 543 | + | sess.snapshotMu.Unlock() | |
| 544 | + | }) | |
| 545 | + | } | |
| 546 | + | } | |
| 547 | + | ||
| 548 | + | func TestSnapshotSchedulerImmediateEligibilityAndStaleCapturesRemainDirty(t *testing.T) { | |
| 549 | + | now := time.Unix(100, 0) | |
| 550 | + | clock := newDeterministicSnapshotClock(now) | |
| 551 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock) | |
| 552 | + | WithSnapshotRepository(portsmocks.NewMockSnapshotRepository(t), nil)(d) | |
| 553 | + | ||
| 554 | + | t.Run("no prior attempt is immediately eligible", func(t *testing.T) { | |
| 555 | + | sess := newSnapshotTestSession(t, "immediate", false, "/work") | |
| 556 | + | sess.snapshotWake = d.snapshotWake | |
| 557 | + | sess.snapDirty.Store(true) | |
| 558 | + | d.sessions[sess.id] = sess | |
| 559 | + | ||
| 560 | + | require.Equal(t, snapshotInterval, d.scheduleEligibleRepositorySnapshots()) | |
| 561 | + | sess.snapshotMu.Lock() | |
| 562 | + | require.True(t, sess.snapshotAttempted) | |
| 563 | + | require.Equal(t, uint64(0), sess.snapshotCapturedGeneration) | |
| 564 | + | require.Equal(t, now.Add(snapshotInterval), sess.snapshotNextEligibleAt) | |
| 565 | + | sess.snapshotMu.Unlock() | |
| 566 | + | }) | |
| 567 | + | ||
| 568 | + | t.Run("queued and in-flight stale successes cannot clear newer generation", func(t *testing.T) { | |
| 569 | + | for _, state := range []struct { | |
| 570 | + | name string | |
| 571 | + | set func(*session, *snapshotCapture) | |
| 572 | + | }{ | |
| 573 | + | {name: "queued", set: func(sess *session, capture *snapshotCapture) { sess.snapshotQueuedCapture = capture }}, | |
| 574 | + | {name: "in flight", set: func(sess *session, capture *snapshotCapture) { sess.snapshotInFlightCapture = capture }}, | |
| 575 | + | } { | |
| 576 | + | t.Run(state.name, func(t *testing.T) { | |
| 577 | + | sess := newSnapshotTestSession(t, state.name, false, "/work") | |
| 578 | + | sess.snapshotGeneration = 1 | |
| 579 | + | sess.snapDirty.Store(true) | |
| 580 | + | capture := &snapshotCapture{session: sess, generation: 1, attemptKind: snapshotAttemptRoutine} | |
| 581 | + | sess.snapshotPendingCaptures = 1 | |
| 582 | + | sess.snapshotPending = true | |
| 583 | + | state.set(sess, capture) | |
| 584 | + | ||
| 585 | + | markSnapshotDirty(sess) | |
| 586 | + | d.finishSnapshotCapture(capture, true) | |
| 587 | + | ||
| 588 | + | require.True(t, sess.snapDirty.Load()) | |
| 589 | + | require.Equal(t, uint64(1), sess.snapshotPublishedGeneration) | |
| 590 | + | require.Equal(t, now.Add(snapshotInterval), sess.snapshotNextEligibleAt) | |
| 591 | + | }) | |
| 592 | + | } | |
| 593 | + | }) | |
| 594 | + | } | |
| 595 | + | ||
| 596 | + | func TestSnapshotWorkerQueueIsBoundedAndForcedCapturesCoalesce(t *testing.T) { | |
| 597 | + | now := time.Unix(100, 0) | |
| 598 | + | clock := newDeterministicSnapshotClock(now) | |
| 599 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock) | |
| 600 | + | repository := portsmocks.NewMockSnapshotRepository(t) | |
| 601 | + | WithSnapshotRepository(repository, nil)(d) | |
| 602 | + | startSnapshotEncodeWorker(t, d) | |
| 603 | + | ||
| 604 | + | started := make(chan struct{}) | |
| 605 | + | release := make(chan struct{}) | |
| 606 | + | publications := make(chan ports.SnapshotPublication, 3) | |
| 607 | + | var blockFirst sync.Once | |
| 608 | + | repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error { | |
| 609 | + | publications <- publication | |
| 610 | + | if publication.Name == "in-flight" { | |
| 611 | + | blockFirst.Do(func() { | |
| 612 | + | close(started) | |
| 613 | + | <-release | |
| 614 | + | }) | |
| 615 | + | } | |
| 616 | + | return nil | |
| 617 | + | }).Times(3) | |
| 618 | + | ||
| 619 | + | inFlight := newSnapshotTestSession(t, "in-flight", false, "/work") | |
| 620 | + | markSnapshotDirty(inFlight) | |
| 621 | + | require.True(t, d.scheduleSnapshot(inFlight)) | |
| 622 | + | <-started | |
| 623 | + | ||
| 624 | + | queued := newSnapshotTestSession(t, "queued", false, "/work") | |
| 625 | + | markSnapshotDirty(queued) | |
| 626 | + | require.True(t, d.scheduleSnapshot(queued)) | |
| 627 | + | ||
| 628 | + | d.snapshotWorkerMu.Lock() | |
| 629 | + | require.NotNil(t, d.snapshotWorkerInFlight) | |
| 630 | + | require.Len(t, d.snapshotJobs, snapshotQueueCapacity) | |
| 631 | + | d.snapshotWorkerMu.Unlock() | |
| 632 | + | ||
| 633 | + | saturated := newSnapshotTestSession(t, "saturated", false, "/work") | |
| 634 | + | markSnapshotDirty(saturated) | |
| 635 | + | require.False(t, d.scheduleSnapshot(saturated), "a saturated queue must not block the producer") | |
| 636 | + | require.True(t, saturated.snapDirty.Load(), "the rejected capture remains retryable") | |
| 637 | + | saturated.snapshotMu.Lock() | |
| 638 | + | require.Equal(t, now.Add(snapshotInterval), saturated.snapshotNextEligibleAt) | |
| 639 | + | saturated.snapshotMu.Unlock() | |
| 640 | + | ||
| 641 | + | // A forced request bypasses the routine rate limit and repeated requests | |
| 642 | + | // coalesce to one successor after the in-flight capture completes. | |
| 643 | + | markSnapshotDirty(inFlight) | |
| 644 | + | require.True(t, d.scheduleFinalSnapshot(inFlight)) | |
| 645 | + | markSnapshotDirty(inFlight) | |
| 646 | + | require.True(t, d.scheduleFinalSnapshot(inFlight)) | |
| 647 | + | close(release) | |
| 648 | + | ||
| 649 | + | gotGenerations := []uint64{(<-publications).Generation, (<-publications).Generation, (<-publications).Generation} | |
| 650 | + | require.ElementsMatch(t, []uint64{1, 1, 3}, gotGenerations, "one forced successor bypasses the routine limit while queued work remains bounded") | |
| 651 | + | awaitSnapshotCleanSignal(inFlight) | |
| 652 | + | } | |
| 653 | + | ||
| 654 | + | func TestSnapshotRepositorySaverUsesEarliestDeadlineAndRecomputes(t *testing.T) { | |
| 655 | + | now := time.Unix(100, 0) | |
| 656 | + | clock := newDeterministicSnapshotClock(now) | |
| 657 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock) | |
| 658 | + | repository := portsmocks.NewMockSnapshotRepository(t) | |
| 659 | + | WithSnapshotRepository(repository, nil)(d) | |
| 660 | + | startSnapshotEncodeWorker(t, d) | |
| 661 | + | ||
| 662 | + | published := make(chan ports.SnapshotPublication, 1) | |
| 663 | + | repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error { | |
| 664 | + | published <- publication | |
| 665 | + | return nil | |
| 666 | + | }).Once() | |
| 667 | + | ||
| 668 | + | later := newSnapshotTestSession(t, "later", false, "/work") | |
| 669 | + | later.snapshotWake = d.snapshotWake | |
| 670 | + | later.snapDirty.Store(true) | |
| 671 | + | later.snapshotNextEligibleAt = now.Add(10 * time.Minute) | |
| 672 | + | d.sessions[later.id] = later | |
| 673 | + | ||
| 674 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 675 | + | done := make(chan struct{}) | |
| 676 | + | go func() { | |
| 677 | + | d.snapshotRepositorySaver(ctx) | |
| 678 | + | close(done) | |
| 679 | + | }() | |
| 680 | + | timer := clock.timer() | |
| 681 | + | require.Equal(t, 10*time.Minute, <-timer.resets) | |
| 682 | + | ||
| 683 | + | earlier := newSnapshotTestSession(t, "earlier", false, "/work") | |
| 684 | + | earlier.snapshotWake = d.snapshotWake | |
| 685 | + | earlier.snapDirty.Store(true) | |
| 686 | + | earlier.snapshotNextEligibleAt = now.Add(5 * time.Minute) | |
| 687 | + | d.sessions[earlier.id] = earlier | |
| 688 | + | d.snapshotWake <- struct{}{} | |
| 689 | + | require.Equal(t, 5*time.Minute, <-timer.resets) | |
| 690 | + | ||
| 691 | + | clock.advance(now.Add(5 * time.Minute)) | |
| 692 | + | timer.fire() | |
| 693 | + | // The saver first recomputes immediately after queueing the due capture. | |
| 694 | + | require.Equal(t, 5*time.Minute, <-timer.resets) | |
| 695 | + | publication := <-published | |
| 696 | + | require.Equal(t, "earlier", publication.Name, "timer fires exactly at the earliest eligibility") | |
| 697 | + | for { | |
| 698 | + | earlier.snapshotMu.Lock() | |
| 699 | + | clean := !earlier.snapDirty.Load() | |
| 700 | + | changed := earlier.snapshotChangeLocked() | |
| 701 | + | earlier.snapshotMu.Unlock() | |
| 702 | + | if clean { | |
| 703 | + | break | |
| 704 | + | } | |
| 705 | + | <-changed | |
| 706 | + | } | |
| 707 | + | // The later dirty session is still the next deadline after completion. | |
| 708 | + | require.Equal(t, 5*time.Minute, <-timer.resets) | |
| 709 | + | ||
| 710 | + | delete(d.sessions, later.id) | |
| 711 | + | d.snapshotWake <- struct{}{} | |
| 712 | + | require.Equal(t, 24*time.Hour, <-timer.resets) | |
| 713 | + | ||
| 714 | + | cancel() | |
| 715 | + | <-done | |
| 716 | + | } | |
| @@ -451,3 +451,266 @@ func (t *snapshotDeadlineTimer) C() <-chan time.Time { return t.ch } | |||||
| 451 | func (*snapshotDeadlineTimer) Reset(time.Duration) bool { return true } | 451 | func (*snapshotDeadlineTimer) Reset(time.Duration) bool { return true } | ||
| 452 | func (*snapshotDeadlineTimer) Stop() bool { return true } | 452 | func (*snapshotDeadlineTimer) Stop() bool { return true } | ||
| 453 | func (t *snapshotDeadlineTimer) fire() { t.ch <- time.Unix(101, 0) } | 453 | func (t *snapshotDeadlineTimer) fire() { t.ch <- time.Unix(101, 0) } | ||
| 454 | + | ||||
| 455 | + | // deterministicSnapshotClock records timer resets and advances only when a | |||
| 456 | + | // test explicitly fires its timer. It deliberately has no wall-clock behavior. | |||
| 457 | + | type deterministicSnapshotClock struct { | |||
| 458 | + | mu sync.Mutex | |||
| 459 | + | now time.Time | |||
| 460 | + | timers chan *deterministicSnapshotTimer | |||
| 461 | + | } | |||
| 462 | + | ||||
| 463 | + | func newDeterministicSnapshotClock(now time.Time) *deterministicSnapshotClock { | |||
| 464 | + | return &deterministicSnapshotClock{now: now, timers: make(chan *deterministicSnapshotTimer, 1)} | |||
| 465 | + | } | |||
| 466 | + | ||||
| 467 | + | func (c *deterministicSnapshotClock) Now() time.Time { | |||
| 468 | + | c.mu.Lock() | |||
| 469 | + | defer c.mu.Unlock() | |||
| 470 | + | return c.now | |||
| 471 | + | } | |||
| 472 | + | ||||
| 473 | + | func (c *deterministicSnapshotClock) advance(now time.Time) { | |||
| 474 | + | c.mu.Lock() | |||
| 475 | + | c.now = now | |||
| 476 | + | c.mu.Unlock() | |||
| 477 | + | } | |||
| 478 | + | ||||
| 479 | + | func (c *deterministicSnapshotClock) NewTimer(time.Duration) ports.Timer { | |||
| 480 | + | timer := &deterministicSnapshotTimer{clock: c, ch: make(chan time.Time, 1), resets: make(chan time.Duration, 16)} | |||
| 481 | + | c.timers <- timer | |||
| 482 | + | return timer | |||
| 483 | + | } | |||
| 484 | + | ||||
| 485 | + | func (c *deterministicSnapshotClock) timer() *deterministicSnapshotTimer { return <-c.timers } | |||
| 486 | + | ||||
| 487 | + | type deterministicSnapshotTimer struct { | |||
| 488 | + | clock *deterministicSnapshotClock | |||
| 489 | + | ch chan time.Time | |||
| 490 | + | resets chan time.Duration | |||
| 491 | + | } | |||
| 492 | + | ||||
| 493 | + | func (t *deterministicSnapshotTimer) C() <-chan time.Time { return t.ch } | |||
| 494 | + | func (t *deterministicSnapshotTimer) Reset(delay time.Duration) bool { | |||
| 495 | + | t.resets <- delay | |||
| 496 | + | return true | |||
| 497 | + | } | |||
| 498 | + | func (*deterministicSnapshotTimer) Stop() bool { return true } | |||
| 499 | + | func (t *deterministicSnapshotTimer) fire() { t.ch <- t.clock.Now() } | |||
| 500 | + | ||||
| 501 | + | func awaitSnapshotCleanSignal(sess *session) { | |||
| 502 | + | for { | |||
| 503 | + | sess.snapshotMu.Lock() | |||
| 504 | + | clean := !sess.snapDirty.Load() | |||
| 505 | + | changed := sess.snapshotChangeLocked() | |||
| 506 | + | sess.snapshotMu.Unlock() | |||
| 507 | + | if clean { | |||
| 508 | + | return | |||
| 509 | + | } | |||
| 510 | + | <-changed | |||
| 511 | + | } | |||
| 512 | + | } | |||
| 513 | + | ||||
| 514 | + | func TestSnapshotCompletionSchedulesRoutineRetriesFromCompletion(t *testing.T) { | |||
| 515 | + | now := time.Unix(100, 0) | |||
| 516 | + | clock := newDeterministicSnapshotClock(now) | |||
| 517 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock) | |||
| 518 | + | repository := portsmocks.NewMockSnapshotRepository(t) | |||
| 519 | + | WithSnapshotRepository(repository, nil)(d) | |||
| 520 | + | ||||
| 521 | + | for _, tt := range []struct { | |||
| 522 | + | name string | |||
| 523 | + | kind snapshotAttemptKind | |||
| 524 | + | succeeded bool | |||
| 525 | + | existingEligible time.Time | |||
| 526 | + | wantDirty bool | |||
| 527 | + | wantEligible time.Time | |||
| 528 | + | }{ | |||
| 529 | + | {name: "routine success", kind: snapshotAttemptRoutine, succeeded: true, wantEligible: now.Add(snapshotInterval)}, | |||
| 530 | + | {name: "routine failure", kind: snapshotAttemptRoutine, wantDirty: true, wantEligible: now.Add(snapshotInterval)}, | |||
| 531 | + | {name: "forced completion preserves routine eligibility", kind: snapshotAttemptForced, succeeded: true, existingEligible: now.Add(time.Minute), wantEligible: now.Add(time.Minute)}, | |||
| 532 | + | } { | |||
| 533 | + | t.Run(tt.name, func(t *testing.T) { | |||
| 534 | + | sess := newSnapshotTestSession(t, "work", false, "/work") | |||
| 535 | + | sess.snapshotGeneration = 1 | |||
| 536 | + | sess.snapshotNextEligibleAt = tt.existingEligible | |||
| 537 | + | sess.snapDirty.Store(true) | |||
| 538 | + | d.finishSnapshotCapture(&snapshotCapture{session: sess, generation: 1, attemptKind: tt.kind}, tt.succeeded) | |||
| 539 | + | ||||
| 540 | + | require.Equal(t, tt.wantDirty, sess.snapDirty.Load()) | |||
| 541 | + | sess.snapshotMu.Lock() | |||
| 542 | + | require.Equal(t, tt.wantEligible, sess.snapshotNextEligibleAt) | |||
| 543 | + | sess.snapshotMu.Unlock() | |||
| 544 | + | }) | |||
| 545 | + | } | |||
| 546 | + | } | |||
| 547 | + | ||||
| 548 | + | func TestSnapshotSchedulerImmediateEligibilityAndStaleCapturesRemainDirty(t *testing.T) { | |||
| 549 | + | now := time.Unix(100, 0) | |||
| 550 | + | clock := newDeterministicSnapshotClock(now) | |||
| 551 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock) | |||
| 552 | + | WithSnapshotRepository(portsmocks.NewMockSnapshotRepository(t), nil)(d) | |||
| 553 | + | ||||
| 554 | + | t.Run("no prior attempt is immediately eligible", func(t *testing.T) { | |||
| 555 | + | sess := newSnapshotTestSession(t, "immediate", false, "/work") | |||
| 556 | + | sess.snapshotWake = d.snapshotWake | |||
| 557 | + | sess.snapDirty.Store(true) | |||
| 558 | + | d.sessions[sess.id] = sess | |||
| 559 | + | ||||
| 560 | + | require.Equal(t, snapshotInterval, d.scheduleEligibleRepositorySnapshots()) | |||
| 561 | + | sess.snapshotMu.Lock() | |||
| 562 | + | require.True(t, sess.snapshotAttempted) | |||
| 563 | + | require.Equal(t, uint64(0), sess.snapshotCapturedGeneration) | |||
| 564 | + | require.Equal(t, now.Add(snapshotInterval), sess.snapshotNextEligibleAt) | |||
| 565 | + | sess.snapshotMu.Unlock() | |||
| 566 | + | }) | |||
| 567 | + | ||||
| 568 | + | t.Run("queued and in-flight stale successes cannot clear newer generation", func(t *testing.T) { | |||
| 569 | + | for _, state := range []struct { | |||
| 570 | + | name string | |||
| 571 | + | set func(*session, *snapshotCapture) | |||
| 572 | + | }{ | |||
| 573 | + | {name: "queued", set: func(sess *session, capture *snapshotCapture) { sess.snapshotQueuedCapture = capture }}, | |||
| 574 | + | {name: "in flight", set: func(sess *session, capture *snapshotCapture) { sess.snapshotInFlightCapture = capture }}, | |||
| 575 | + | } { | |||
| 576 | + | t.Run(state.name, func(t *testing.T) { | |||
| 577 | + | sess := newSnapshotTestSession(t, state.name, false, "/work") | |||
| 578 | + | sess.snapshotGeneration = 1 | |||
| 579 | + | sess.snapDirty.Store(true) | |||
| 580 | + | capture := &snapshotCapture{session: sess, generation: 1, attemptKind: snapshotAttemptRoutine} | |||
| 581 | + | sess.snapshotPendingCaptures = 1 | |||
| 582 | + | sess.snapshotPending = true | |||
| 583 | + | state.set(sess, capture) | |||
| 584 | + | ||||
| 585 | + | markSnapshotDirty(sess) | |||
| 586 | + | d.finishSnapshotCapture(capture, true) | |||
| 587 | + | ||||
| 588 | + | require.True(t, sess.snapDirty.Load()) | |||
| 589 | + | require.Equal(t, uint64(1), sess.snapshotPublishedGeneration) | |||
| 590 | + | require.Equal(t, now.Add(snapshotInterval), sess.snapshotNextEligibleAt) | |||
| 591 | + | }) | |||
| 592 | + | } | |||
| 593 | + | }) | |||
| 594 | + | } | |||
| 595 | + | ||||
| 596 | + | func TestSnapshotWorkerQueueIsBoundedAndForcedCapturesCoalesce(t *testing.T) { | |||
| 597 | + | now := time.Unix(100, 0) | |||
| 598 | + | clock := newDeterministicSnapshotClock(now) | |||
| 599 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock) | |||
| 600 | + | repository := portsmocks.NewMockSnapshotRepository(t) | |||
| 601 | + | WithSnapshotRepository(repository, nil)(d) | |||
| 602 | + | startSnapshotEncodeWorker(t, d) | |||
| 603 | + | ||||
| 604 | + | started := make(chan struct{}) | |||
| 605 | + | release := make(chan struct{}) | |||
| 606 | + | publications := make(chan ports.SnapshotPublication, 3) | |||
| 607 | + | var blockFirst sync.Once | |||
| 608 | + | repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error { | |||
| 609 | + | publications <- publication | |||
| 610 | + | if publication.Name == "in-flight" { | |||
| 611 | + | blockFirst.Do(func() { | |||
| 612 | + | close(started) | |||
| 613 | + | <-release | |||
| 614 | + | }) | |||
| 615 | + | } | |||
| 616 | + | return nil | |||
| 617 | + | }).Times(3) | |||
| 618 | + | ||||
| 619 | + | inFlight := newSnapshotTestSession(t, "in-flight", false, "/work") | |||
| 620 | + | markSnapshotDirty(inFlight) | |||
| 621 | + | require.True(t, d.scheduleSnapshot(inFlight)) | |||
| 622 | + | <-started | |||
| 623 | + | ||||
| 624 | + | queued := newSnapshotTestSession(t, "queued", false, "/work") | |||
| 625 | + | markSnapshotDirty(queued) | |||
| 626 | + | require.True(t, d.scheduleSnapshot(queued)) | |||
| 627 | + | ||||
| 628 | + | d.snapshotWorkerMu.Lock() | |||
| 629 | + | require.NotNil(t, d.snapshotWorkerInFlight) | |||
| 630 | + | require.Len(t, d.snapshotJobs, snapshotQueueCapacity) | |||
| 631 | + | d.snapshotWorkerMu.Unlock() | |||
| 632 | + | ||||
| 633 | + | saturated := newSnapshotTestSession(t, "saturated", false, "/work") | |||
| 634 | + | markSnapshotDirty(saturated) | |||
| 635 | + | require.False(t, d.scheduleSnapshot(saturated), "a saturated queue must not block the producer") | |||
| 636 | + | require.True(t, saturated.snapDirty.Load(), "the rejected capture remains retryable") | |||
| 637 | + | saturated.snapshotMu.Lock() | |||
| 638 | + | require.Equal(t, now.Add(snapshotInterval), saturated.snapshotNextEligibleAt) | |||
| 639 | + | saturated.snapshotMu.Unlock() | |||
| 640 | + | ||||
| 641 | + | // A forced request bypasses the routine rate limit and repeated requests | |||
| 642 | + | // coalesce to one successor after the in-flight capture completes. | |||
| 643 | + | markSnapshotDirty(inFlight) | |||
| 644 | + | require.True(t, d.scheduleFinalSnapshot(inFlight)) | |||
| 645 | + | markSnapshotDirty(inFlight) | |||
| 646 | + | require.True(t, d.scheduleFinalSnapshot(inFlight)) | |||
| 647 | + | close(release) | |||
| 648 | + | ||||
| 649 | + | gotGenerations := []uint64{(<-publications).Generation, (<-publications).Generation, (<-publications).Generation} | |||
| 650 | + | require.ElementsMatch(t, []uint64{1, 1, 3}, gotGenerations, "one forced successor bypasses the routine limit while queued work remains bounded") | |||
| 651 | + | awaitSnapshotCleanSignal(inFlight) | |||
| 652 | + | } | |||
| 653 | + | ||||
| 654 | + | func TestSnapshotRepositorySaverUsesEarliestDeadlineAndRecomputes(t *testing.T) { | |||
| 655 | + | now := time.Unix(100, 0) | |||
| 656 | + | clock := newDeterministicSnapshotClock(now) | |||
| 657 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock) | |||
| 658 | + | repository := portsmocks.NewMockSnapshotRepository(t) | |||
| 659 | + | WithSnapshotRepository(repository, nil)(d) | |||
| 660 | + | startSnapshotEncodeWorker(t, d) | |||
| 661 | + | ||||
| 662 | + | published := make(chan ports.SnapshotPublication, 1) | |||
| 663 | + | repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error { | |||
| 664 | + | published <- publication | |||
| 665 | + | return nil | |||
| 666 | + | }).Once() | |||
| 667 | + | ||||
| 668 | + | later := newSnapshotTestSession(t, "later", false, "/work") | |||
| 669 | + | later.snapshotWake = d.snapshotWake | |||
| 670 | + | later.snapDirty.Store(true) | |||
| 671 | + | later.snapshotNextEligibleAt = now.Add(10 * time.Minute) | |||
| 672 | + | d.sessions[later.id] = later | |||
| 673 | + | ||||
| 674 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 675 | + | done := make(chan struct{}) | |||
| 676 | + | go func() { | |||
| 677 | + | d.snapshotRepositorySaver(ctx) | |||
| 678 | + | close(done) | |||
| 679 | + | }() | |||
| 680 | + | timer := clock.timer() | |||
| 681 | + | require.Equal(t, 10*time.Minute, <-timer.resets) | |||
| 682 | + | ||||
| 683 | + | earlier := newSnapshotTestSession(t, "earlier", false, "/work") | |||
| 684 | + | earlier.snapshotWake = d.snapshotWake | |||
| 685 | + | earlier.snapDirty.Store(true) | |||
| 686 | + | earlier.snapshotNextEligibleAt = now.Add(5 * time.Minute) | |||
| 687 | + | d.sessions[earlier.id] = earlier | |||
| 688 | + | d.snapshotWake <- struct{}{} | |||
| 689 | + | require.Equal(t, 5*time.Minute, <-timer.resets) | |||
| 690 | + | ||||
| 691 | + | clock.advance(now.Add(5 * time.Minute)) | |||
| 692 | + | timer.fire() | |||
| 693 | + | // The saver first recomputes immediately after queueing the due capture. | |||
| 694 | + | require.Equal(t, 5*time.Minute, <-timer.resets) | |||
| 695 | + | publication := <-published | |||
| 696 | + | require.Equal(t, "earlier", publication.Name, "timer fires exactly at the earliest eligibility") | |||
| 697 | + | for { | |||
| 698 | + | earlier.snapshotMu.Lock() | |||
| 699 | + | clean := !earlier.snapDirty.Load() | |||
| 700 | + | changed := earlier.snapshotChangeLocked() | |||
| 701 | + | earlier.snapshotMu.Unlock() | |||
| 702 | + | if clean { | |||
| 703 | + | break | |||
| 704 | + | } | |||
| 705 | + | <-changed | |||
| 706 | + | } | |||
| 707 | + | // The later dirty session is still the next deadline after completion. | |||
| 708 | + | require.Equal(t, 5*time.Minute, <-timer.resets) | |||
| 709 | + | ||||
| 710 | + | delete(d.sessions, later.id) | |||
| 711 | + | d.snapshotWake <- struct{}{} | |||
| 712 | + | require.Equal(t, 24*time.Hour, <-timer.resets) | |||
| 713 | + | ||||
| 714 | + | cancel() | |||
| 715 | + | <-done | |||
| 716 | + | } | |||