| 1 | package schedule |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "fmt" |
| 6 | "github.com/carlosframework/platform/internal/hostwrite" |
| 7 | "github.com/carlosframework/platform/internal/registry" |
| 8 | "testing" |
| 9 | "time" |
| 10 | ) |
| 11 | |
| 12 | type protectedTimerWork struct { |
| 13 | t *testing.T |
| 14 | reg *registry.Registry |
| 15 | held bool |
| 16 | calls int |
| 17 | } |
| 18 | |
| 19 | func (w *protectedTimerWork) With(_ context.Context, _ hostwrite.Identity, f func() error) error { |
| 20 | if w.held { |
| 21 | w.t.Fatal("recursive host exclusion") |
| 22 | } |
| 23 | w.held = true |
| 24 | defer func() { w.held = false }() |
| 25 | return f() |
| 26 | } |
| 27 | func (w *protectedTimerWork) Select(context.Context, string, string) (hostwrite.Work, error) { |
| 28 | return w, nil |
| 29 | } |
| 30 | func (w *protectedTimerWork) MutateTimer(_ context.Context, row registry.Timer, expected string, remove bool) (*registry.Timer, error) { |
| 31 | if w.held { |
| 32 | w.t.Fatal("root timer RPC inside host exclusion") |
| 33 | } |
| 34 | w.calls++ |
| 35 | if remove { |
| 36 | return nil, w.reg.DeleteTimerOwnedRegistration(row) |
| 37 | } |
| 38 | rt, err := w.reg.Get(row.Host) |
| 39 | if err != nil { |
| 40 | return nil, err |
| 41 | } |
| 42 | row.ClaimOwnership = rt.ClaimOwnership |
| 43 | row.Registration = fmt.Sprint("root-", w.calls) |
| 44 | row.ClaimRowID = row.Registration |
| 45 | return w.reg.MaterializeTimerRegistration(row, expected) |
| 46 | } |
| 47 | func TestPlan14TimerProducerTransactions(t *testing.T) { |
| 48 | t.Setenv("CARLOS_HOST_CLAIM", "1") |
| 49 | reg, _ := convergeFixture(t) |
| 50 | host := "a.bab.oncarlos.com" |
| 51 | rt, err := reg.Get(host) |
| 52 | if err != nil { |
| 53 | t.Fatal(err) |
| 54 | } |
| 55 | owner := registry.ClaimOwnership{ClaimIncarnation: "inc", ClaimStamp: 1, ClaimOwner: "owner", ClaimRowID: "route"} |
| 56 | if err := reg.AdoptClaimRoute(t.Context(), *rt, owner); err != nil { |
| 57 | t.Fatal(err) |
| 58 | } |
| 59 | rt.ClaimOwnership = owner |
| 60 | w := &protectedTimerWork{t: t, reg: reg} |
| 61 | writes := hostwrite.Config{Enabled: true, Selector: w} |
| 62 | s := ClaimTimers{TimerStore: reg, Writes: writes, Context: t.Context()} |
| 63 | row := registry.Timer{Host: host, Name: "once", Kind: KindOnce, Spec: "at 2000", Path: "/once", NextAt: 2000} |
| 64 | if err := s.UpsertTimer(row); err != nil { |
| 65 | t.Fatal(err) |
| 66 | } |
| 67 | first, _ := reg.GetTimer(host, row.Name) |
| 68 | if err := s.UpsertTimer(row); err != nil { |
| 69 | t.Fatal(err) |
| 70 | } |
| 71 | second, _ := reg.GetTimer(host, row.Name) |
| 72 | if first.Registration == second.Registration { |
| 73 | t.Fatal("once timer was not re-registered") |
| 74 | } |
| 75 | if err := s.DeleteTimer(host, row.Name); err != nil { |
| 76 | t.Fatal(err) |
| 77 | } |
| 78 | ctx := context.WithValue(t.Context(), timerWorkKey{}, timerWork{ctx: t.Context(), work: w}) |
| 79 | c := NewConverger(reg, nil, "box", nil) |
| 80 | c.HostWrites = writes |
| 81 | d := Declaration{Schedules: []Schedule{{Name: "tick", Every: "1h", Path: "/tick"}}} |
| 82 | if _, err := c.convergeHost(ctx, *rt, d, "", time.Unix(1000, 0)); err != nil { |
| 83 | t.Fatal(err) |
| 84 | } |
| 85 | original, _ := reg.GetTimer(host, "tick") |
| 86 | if _, err := c.convergeHost(ctx, *rt, d, "tick", time.Unix(1100, 0)); err != nil { |
| 87 | t.Fatal(err) |
| 88 | } |
| 89 | fired, _ := reg.GetTimer(host, "tick") |
| 90 | if original.Registration == fired.Registration || fired.NextAt != 1100 { |
| 91 | t.Fatal("run-now skipped root mutation", fired) |
| 92 | } |
| 93 | if _, err := c.convergeHost(ctx, *rt, Declaration{}, "", time.Unix(1200, 0)); err != nil { |
| 94 | t.Fatal(err) |
| 95 | } |
| 96 | if w.calls != 6 { |
| 97 | t.Fatalf("producers used %d root mutations, want 6", w.calls) |
| 98 | } |
| 99 | selected := SelectedTimers{Store: reg, Work: w, Context: t.Context(), Host: host} |
| 100 | if err := selected.ReplaceTimers(host, []registry.Timer{row}); err == nil || w.calls != 6 { |
| 101 | t.Fatal("import minted missing registration", err, w.calls) |
| 102 | } |
| 103 | row.Registration = "snapshot" |
| 104 | if err := selected.ReplaceTimers(host, []registry.Timer{row, row}); err == nil || w.calls != 6 { |
| 105 | t.Fatal("duplicate import mutated rows", err, w.calls) |
| 106 | } |
| 107 | if err := selected.ReplaceTimers(host, []registry.Timer{row}); err != nil { |
| 108 | t.Fatal(err) |
| 109 | } |
| 110 | if err := selected.ReplaceTimers(host, nil); err != nil { |
| 111 | t.Fatal(err) |
| 112 | } |
| 113 | if w.calls != 8 { |
| 114 | t.Fatal("landing bypassed root mutations", w.calls) |
| 115 | } |
| 116 | if err := s.UpsertTimer(row); err != nil { |
| 117 | t.Fatal(err) |
| 118 | } |
| 119 | old, _ := reg.GetTimer(host, row.Name) |
| 120 | if err := deleteSelectedTimer(ctx, reg, *old); err != nil { |
| 121 | t.Fatal(err) |
| 122 | } |
| 123 | if w.calls != 10 { |
| 124 | t.Fatal("runner prune bypassed root mutation", w.calls) |
| 125 | } |
| 126 | } |
| 127 | |
amadan