| index 4143f6821..f421b4610 100644 |
| --- a/internal/host/movescheckpoint_test.go |
| +++ b/internal/host/movescheckpoint_test.go |
| @@ -13,19 +13,28 @@ import ( |
| "time" |
| |
| "github.com/carlosframework/platform/internal/activator" |
| + "github.com/carlosframework/platform/internal/lease" |
| "github.com/carlosframework/platform/internal/provision" |
| "github.com/carlosframework/platform/internal/usage" |
| ) |
| |
| // checkpointS3 is enough of S3 for a leased litestream sync: an empty |
| -// listing, absent objects, and writes that succeed. It records each write |
| -// by path. |
| +// listing, absent objects, and writes that succeed. It records every |
| +// physical request so the ledger can be reconciled with the store. |
| type checkpointS3 struct { |
| - mu sync.Mutex |
| - puts []string |
| + mu sync.Mutex |
| + requests []checkpointRequest |
| +} |
| + |
| +type checkpointRequest struct { |
| + Method string |
| + Path string |
| } |
| |
| func (f *checkpointS3) serve(w http.ResponseWriter, r *http.Request) { |
| + f.mu.Lock() |
| + f.requests = append(f.requests, checkpointRequest{Method: r.Method, Path: r.URL.Path}) |
| + f.mu.Unlock() |
| _, _ = io.Copy(io.Discard, r.Body) |
| w.Header().Set("Content-Type", "application/xml") |
| switch { |
| @@ -35,9 +44,6 @@ func (f *checkpointS3) serve(w http.ResponseWriter, r *http.Request) { |
| w.WriteHeader(http.StatusNotFound) |
| _, _ = io.WriteString(w, `<?xml version="1.0" encoding="UTF-8"?><Error><Code>NoSuchKey</Code><Message>no such key</Message></Error>`) |
| case r.Method == http.MethodPut: |
| - f.mu.Lock() |
| - f.puts = append(f.puts, r.URL.Path) |
| - f.mu.Unlock() |
| w.Header().Set("ETag", `"etag"`) |
| case r.Method == http.MethodDelete: |
| w.WriteHeader(http.StatusNoContent) |
| @@ -50,14 +56,125 @@ func (f *checkpointS3) replicaPuts() int { |
| f.mu.Lock() |
| defer f.mu.Unlock() |
| n := 0 |
| - for _, p := range f.puts { |
| - if !strings.HasSuffix(p, "/lock.json") { |
| + for _, request := range f.requests { |
| + if request.Method == http.MethodPut && !strings.HasSuffix(request.Path, "/lock.json") { |
| n++ |
| } |
| } |
| return n |
| } |
| |
| +func (f *checkpointS3) requestCounts() (leaseRequests, replicaRequests int, requests []checkpointRequest) { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + requests = append(requests, f.requests...) |
| + for _, request := range requests { |
| + if strings.HasSuffix(request.Path, "/lock.json") { |
| + leaseRequests++ |
| + } else { |
| + replicaRequests++ |
| + } |
| + } |
| + return leaseRequests, replicaRequests, requests |
| +} |
| + |
| +type checkpointLedgerEvidence struct { |
| + leaseAttempts int64 |
| + checkpointAttempts int64 |
| + checkpointUnknown int64 |
| + checkpointAnswered int64 |
| +} |
| + |
| +func checkpointEvidence(t *testing.T, content usage.DurabilityContent, rec provision.Record) checkpointLedgerEvidence { |
| + t.Helper() |
| + var evidence checkpointLedgerEvidence |
| + for _, rows := range content.Hours { |
| + for _, row := range rows { |
| + if row.Account != rec.Account || row.App != rec.App || row.Host != rec.Host { |
| + t.Fatalf("a checkpoint request lost its owner: %+v", row) |
| + } |
| + switch row.Purpose { |
| + case string(usage.PurposeLease): |
| + evidence.leaseAttempts += row.Attempts |
| + case string(usage.PurposeCheckpoint): |
| + evidence.checkpointAttempts += row.Attempts |
| + evidence.checkpointUnknown += row.Unknown |
| + for _, answered := range row.Answered { |
| + evidence.checkpointAnswered += answered |
| + } |
| + default: |
| + t.Fatalf("a checkpoint request used purpose %q: %+v", row.Purpose, row) |
| + } |
| + } |
| + } |
| + return evidence |
| +} |
| + |
| +func checkpointDeficit(requests int, attempts int64) int64 { |
| + deficit := int64(requests) - attempts |
| + if deficit < 0 { |
| + return 0 |
| + } |
| + return deficit |
| +} |
| + |
| +func waitCheckpointEvidence(t *testing.T, ctx context.Context, f *checkpointS3, p *durabilityPublisher, rec provision.Record, requireUnknown bool) (usage.DurabilityContent, checkpointLedgerEvidence) { |
| + t.Helper() |
| + leaseRequests, replicaRequests, requests := f.requestCounts() |
| + for { |
| + content, err := p.ledger.Snapshot(time.Now().UTC().Format("2006-01-02")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + evidence := checkpointEvidence(t, content, rec) |
| + failures := int64(0) |
| + var failureDiagnostic any |
| + if content.ActualRecordingFailures != nil { |
| + failures = *content.ActualRecordingFailures |
| + failureDiagnostic = failures |
| + } |
| + deficit := checkpointDeficit(leaseRequests, evidence.leaseAttempts) + checkpointDeficit(replicaRequests, evidence.checkpointAttempts) |
| + complete := evidence.leaseAttempts > 0 && evidence.checkpointAttempts > 0 && failures >= deficit && content.Held == 0 && content.Barred == 0 |
| + if requireUnknown { |
| + complete = complete && evidence.checkpointUnknown == evidence.checkpointAttempts && evidence.checkpointAnswered == 0 |
| + } |
| + if complete { |
| + return content, evidence |
| + } |
| + select { |
| + case <-ctx.Done(): |
| + t.Fatalf("checkpoint evidence did not settle: %v; actual_recording_failures=%v content=%+v evidence=%+v lease_requests=%d replica_requests=%d requests=%+v", ctx.Err(), failureDiagnostic, content, evidence, leaseRequests, replicaRequests, requests) |
| + case <-time.After(20 * time.Millisecond): |
| + } |
| + } |
| +} |
| + |
| +type blockingCheckpointRecorder struct { |
| + lease.RequestRecorder |
| + release chan struct{} |
| + entered chan struct{} |
| + done chan struct{} |
| + once sync.Once |
| +} |
| + |
| +func newBlockingCheckpointRecorder(recorder lease.RequestRecorder) *blockingCheckpointRecorder { |
| + return &blockingCheckpointRecorder{ |
| + RequestRecorder: recorder, |
| + release: make(chan struct{}), |
| + entered: make(chan struct{}, 4), |
| + done: make(chan struct{}, 4), |
| + } |
| +} |
| + |
| +func (r *blockingCheckpointRecorder) Start(operation string, at time.Time) (uint64, error) { |
| + r.entered <- struct{}{} |
| + defer func() { r.done <- struct{}{} }() |
| + <-r.release |
| + return r.RequestRecorder.Start(operation, at) |
| +} |
| + |
| +func (r *blockingCheckpointRecorder) unblock() { r.once.Do(func() { close(r.release) }) } |
| + |
| func checkpointFixture(t *testing.T) (*checkpointS3, *durabilityPublisher, string, provision.Record) { |
| t.Helper() |
| f := &checkpointS3{} |
| @@ -124,34 +241,81 @@ func TestMovesPassCheckpointIsMeteredToTheRouteOwner(t *testing.T) { |
| if p.bindingFailed.Load() { |
| t.Fatal("a metered checkpoint filed an escape") |
| } |
| - deadline := time.Now().Add(5 * time.Second) |
| - for { |
| - c, err := p.ledger.Snapshot(time.Now().UTC().Format("2006-01-02")) |
| + waitCheckpointEvidence(t, ctx, f, p, rec, false) |
| +} |
| + |
| +func TestMovesPassCheckpointRecordsSaturationAsBoundedIncomplete(t *testing.T) { |
| + f, p, dbPath, rec := checkpointFixture(t) |
| + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) |
| + defer cancel() |
| + |
| + attr := usage.ReplicationAttribution{Account: rec.Account, App: rec.App, Host: rec.Host, Purpose: usage.PurposeCheckpoint} |
| + client, err := p.ledger.Bind(attr) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + blocked := newBlockingCheckpointRecorder(&durabilityRecorder{client}) |
| + t.Cleanup(blocked.unblock) |
| + ready := make(chan struct{}) |
| + close(ready) |
| + p.mu.Lock() |
| + p.bindings[attr] = &durabilityBinding{ready: ready, recorder: blocked} |
| + p.mu.Unlock() |
| + |
| + result := make(chan error, 1) |
| + go func() { result <- armedMovesWorker.Load().cfg.CheckpointFor(ctx, rec, dbPath) }() |
| + for range 4 { |
| + select { |
| + case <-blocked.entered: |
| + case <-ctx.Done(): |
| + t.Fatalf("checkpoint did not fill all recorder slots: %v", ctx.Err()) |
| + } |
| + } |
| + select { |
| + case err := <-result: |
| if err != nil { |
| t.Fatal(err) |
| } |
| - writes := map[string]int64{} |
| - for _, rows := range c.Hours { |
| - for _, row := range rows { |
| - if row.Account != rec.Account || row.App != rec.App || row.Host != rec.Host { |
| - t.Fatalf("a checkpoint request lost its owner: %+v", row) |
| - } |
| - if row.Purpose != string(usage.PurposeCheckpoint) && row.Purpose != string(usage.PurposeLease) { |
| - t.Fatalf("a checkpoint request under purpose %q: %+v", row.Purpose, row) |
| - } |
| - if row.Operation == "PutObject" { |
| - writes[row.Purpose] += row.Attempts |
| - } |
| - } |
| - } |
| - if writes[string(usage.PurposeCheckpoint)] >= int64(f.replicaPuts()) && writes[string(usage.PurposeLease)] > 0 { |
| - return |
| - } |
| - if time.Now().After(deadline) { |
| - t.Fatalf("ledger writes %+v; the store saw %d replica writes", writes, f.replicaPuts()) |
| + case <-ctx.Done(): |
| + t.Fatalf("checkpoint did not finish while recorder starts were blocked: %v", ctx.Err()) |
| + } |
| + if f.replicaPuts() == 0 { |
| + t.Fatal("fixture: the checkpoint wrote nothing to the replica") |
| + } |
| + if p.bindingFailed.Load() { |
| + t.Fatal("a metered checkpoint filed an escape") |
| + } |
| + content, err := p.ledger.Snapshot(time.Now().UTC().Format("2006-01-02")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + evidence := checkpointEvidence(t, content, rec) |
| + _, replicaRequests, requests := f.requestCounts() |
| + failures := int64(0) |
| + var failureDiagnostic any |
| + if content.ActualRecordingFailures != nil { |
| + failures = *content.ActualRecordingFailures |
| + failureDiagnostic = failures |
| + } |
| + // The live meter holds which pin these attempts are transient latch state, |
| + // not the durable uncertainty rows reported by content.Held. Do not require |
| + // the complete-phase zero-held invariant until the workers have drained. |
| + if evidence.checkpointAttempts != 0 || failures < int64(replicaRequests) || content.Barred != 0 { |
| + t.Fatalf("blocked starts did not expose their pending and unrecorded state: actual_recording_failures=%v content=%+v evidence=%+v replica_requests=%d requests=%+v", failureDiagnostic, content, evidence, replicaRequests, requests) |
| + } |
| + |
| + blocked.unblock() |
| + for range 4 { |
| + select { |
| + case <-blocked.done: |
| + case <-ctx.Done(): |
| + t.Fatalf("blocked recorder start did not finish: %v", ctx.Err()) |
| } |
| - time.Sleep(20 * time.Millisecond) |
| } |
| + if err := p.ledger.WaitLate(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + waitCheckpointEvidence(t, ctx, f, p, rec, true) |
| } |
| |
| // With no recorder to bind (the publisher has stopped), the checkpoint |
amadan