| 1 | package schedule |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "github.com/carlosframework/platform/internal/hostwrite" |
| 7 | "github.com/carlosframework/platform/internal/registry" |
| 8 | "net/http" |
| 9 | "sync/atomic" |
| 10 | "testing" |
| 11 | "time" |
| 12 | ) |
| 13 | |
| 14 | type cancelDelivery struct { |
| 15 | started chan struct{} |
| 16 | legacy atomic.Bool |
| 17 | } |
| 18 | |
| 19 | func (d *cancelDelivery) Deliver(context.Context, registry.Route, *http.Request) (int, error) { |
| 20 | d.legacy.Store(true) |
| 21 | return 200, nil |
| 22 | } |
| 23 | func (d *cancelDelivery) DeliverClaim(ctx context.Context, rt registry.Route, req *http.Request, admit func(context.Context) (context.Context, func(), error)) (int, error) { |
| 24 | child, finish, err := admit(ctx) |
| 25 | if err != nil { |
| 26 | return 0, err |
| 27 | } |
| 28 | defer finish() |
| 29 | close(d.started) |
| 30 | <-child.Done() |
| 31 | return 0, child.Err() |
| 32 | } |
| 33 | func TestClaimTimerDeliveryDrainsBeforeBookkeeping(t *testing.T) { |
| 34 | reg, r := runnerFixture(t, &fakeDeliverer{status: 204}) |
| 35 | now := time.Unix(20000, 0) |
| 36 | r.Now = func() time.Time { return now } |
| 37 | valid := &atomic.Bool{} |
| 38 | valid.Store(true) |
| 39 | r.HostWrites = hostwrite.Config{Enabled: true, Selector: lifecycleSelector{valid}} |
| 40 | d := &cancelDelivery{started: make(chan struct{})} |
| 41 | r.Deliver = d |
| 42 | finished := make(chan struct{}) |
| 43 | var stop context.CancelFunc |
| 44 | r.StartDelivery = func(ctx context.Context, w hostwrite.Work, _ registry.Route, _ registry.Timer) (context.Context, func(), error) { |
| 45 | var child context.Context |
| 46 | err := w.With(ctx, hostwrite.Identity{}, func() error { child, stop = context.WithCancel(ctx); return nil }) |
| 47 | return child, func() { close(finished) }, err |
| 48 | } |
| 49 | host := "a.bab.oncarlos.com" |
| 50 | if err := reg.UpsertTimer(registry.Timer{Host: host, Name: "tick", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/tick", NextAt: now.Unix()}); err != nil { |
| 51 | t.Fatal(err) |
| 52 | } |
| 53 | r.Tick(t.Context()) |
| 54 | <-d.started |
| 55 | valid.Store(false) |
| 56 | stop() |
| 57 | r.wait() |
| 58 | select { |
| 59 | case <-finished: |
| 60 | default: |
| 61 | t.Fatal("delivery not joined") |
| 62 | } |
| 63 | row, err := reg.GetTimer(host, "tick") |
| 64 | if err != nil || row.LastAt != 0 || row.NextAt != now.Unix() || d.legacy.Load() { |
| 65 | t.Fatalf("late write or legacy delivery: %+v %v", row, err) |
| 66 | } |
| 67 | r.StartDelivery = nil |
| 68 | valid.Store(true) |
| 69 | ctx := context.WithValue(t.Context(), timerWorkKey{}, timerWork{t.Context(), lifecycleWork{valid}}) |
| 70 | if _, err := r.deliverClaim(ctx, registry.Route{Host: host}, *row, &http.Request{}); !errors.Is(err, hostwrite.ErrUnavailable) { |
| 71 | t.Fatal(err) |
| 72 | } |
| 73 | } |
| 74 | |
amadan