| 1 | package schedule |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "net/http" |
| 7 | "path/filepath" |
| 8 | "sync/atomic" |
| 9 | "testing" |
| 10 | "time" |
| 11 | |
| 12 | "encoding/json" |
| 13 | "github.com/carlosframework/platform/internal/hostwrite" |
| 14 | "github.com/carlosframework/platform/internal/registry" |
| 15 | "github.com/carlosframework/platform/internal/release" |
| 16 | "reflect" |
| 17 | ) |
| 18 | |
| 19 | type lifecycleWork struct{ valid *atomic.Bool } |
| 20 | |
| 21 | func (w lifecycleWork) With(_ context.Context, _ hostwrite.Identity, fn func() error) error { |
| 22 | if !w.valid.Load() { |
| 23 | return errors.New("selection changed") |
| 24 | } |
| 25 | return fn() |
| 26 | } |
| 27 | |
| 28 | type lifecycleSelector struct{ valid *atomic.Bool } |
| 29 | |
| 30 | func (s lifecycleSelector) Select(context.Context, string, string) (hostwrite.Work, error) { |
| 31 | return lifecycleWork{valid: s.valid}, nil |
| 32 | } |
| 33 | |
| 34 | type pausedDelivery struct { |
| 35 | started chan struct{} |
| 36 | resume chan struct{} |
| 37 | } |
| 38 | |
| 39 | func (d pausedDelivery) Deliver(context.Context, registry.Route, *http.Request) (int, error) { |
| 40 | close(d.started) |
| 41 | <-d.resume |
| 42 | return 204, nil |
| 43 | } |
| 44 | |
| 45 | func TestRunnerChangedSelectionAfterDeliveryWritesNoBookkeeping(t *testing.T) { |
| 46 | reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| 47 | if err != nil { |
| 48 | t.Fatal(err) |
| 49 | } |
| 50 | t.Cleanup(func() { reg.Close() }) |
| 51 | host := "app.bab.oncarlos.com" |
| 52 | if err := reg.Upsert(registry.Route{Host: host, Kind: "instance", Account: "bab", App: "app", Status: "ready"}); err != nil { |
| 53 | t.Fatal(err) |
| 54 | } |
| 55 | now := time.Unix(10_000, 0).UTC() |
| 56 | if err := reg.UpsertTimer(registry.Timer{Host: host, Name: "tick", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/tick", NextAt: now.Unix()}); err != nil { |
| 57 | t.Fatal(err) |
| 58 | } |
| 59 | valid := &atomic.Bool{} |
| 60 | valid.Store(true) |
| 61 | delivery := pausedDelivery{started: make(chan struct{}), resume: make(chan struct{})} |
| 62 | events := atomic.Int64{} |
| 63 | r := &Runner{ |
| 64 | StartDelivery: func(ctx context.Context, w hostwrite.Work, _ registry.Route, _ registry.Timer) (context.Context, func(), error) { |
| 65 | err := w.With(ctx, hostwrite.Identity{}, func() error { return nil }) |
| 66 | return ctx, func() {}, err |
| 67 | }, |
| 68 | Reg: reg, Deliver: delivery, Now: func() time.Time { return now }, Timeout: time.Minute, |
| 69 | Token: func(string) (string, error) { return "token", nil }, |
| 70 | Event: func(registry.Route, string, string) { events.Add(1) }, |
| 71 | HostWrites: hostwrite.Config{Enabled: true, Selector: lifecycleSelector{valid: valid}}, |
| 72 | } |
| 73 | r.Tick(t.Context()) |
| 74 | <-delivery.started |
| 75 | valid.Store(false) |
| 76 | close(delivery.resume) |
| 77 | r.wait() |
| 78 | row, err := reg.GetTimer(host, "tick") |
| 79 | if err != nil { |
| 80 | t.Fatal(err) |
| 81 | } |
| 82 | if row.LastAt != 0 || row.NextAt != now.Unix() || events.Load() != 0 { |
| 83 | t.Fatalf("bookkeeping changed after authority loss: row=%+v events=%d", row, events.Load()) |
| 84 | } |
| 85 | } |
| 86 | |
| 87 | func TestEnabledRunnerMissingLifecycleProviderRefusesTimer(t *testing.T) { |
| 88 | d := &fakeDeliverer{status: 204} |
| 89 | reg, r := runnerFixture(t, d) |
| 90 | now := time.Unix(20_000, 0).UTC() |
| 91 | r.Now = func() time.Time { return now } |
| 92 | r.HostWrites = hostwrite.Config{Enabled: true} |
| 93 | if err := reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "tick", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/tick", NextAt: now.Unix()}); err != nil { |
| 94 | t.Fatal(err) |
| 95 | } |
| 96 | r.Tick(t.Context()) |
| 97 | r.wait() |
| 98 | if len(d.calls) != 0 { |
| 99 | t.Fatal("delivery ran without lifecycle provider") |
| 100 | } |
| 101 | } |
| 102 | |
| 103 | type rotatingSelector struct { |
| 104 | current *atomic.Int64 |
| 105 | calls int |
| 106 | } |
| 107 | type capturedEpoch struct { |
| 108 | current *atomic.Int64 |
| 109 | selected int64 |
| 110 | } |
| 111 | |
| 112 | func (s *rotatingSelector) Select(context.Context, string, string) (hostwrite.Work, error) { |
| 113 | s.calls++ |
| 114 | return capturedEpoch{s.current, s.current.Load()}, nil |
| 115 | } |
| 116 | func (w capturedEpoch) With(_ context.Context, _ hostwrite.Identity, fn func() error) error { |
| 117 | if w.current.Load() != w.selected { |
| 118 | return errors.New("installation changed") |
| 119 | } |
| 120 | return fn() |
| 121 | } |
| 122 | |
| 123 | type changeOnRead struct { |
| 124 | release.Store |
| 125 | key string |
| 126 | current *atomic.Int64 |
| 127 | } |
| 128 | |
| 129 | func (s changeOnRead) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 130 | b, e, err := s.Store.Get(ctx, key) |
| 131 | if key == s.key { |
| 132 | s.current.Add(1) |
| 133 | } |
| 134 | return b, e, err |
| 135 | } |
| 136 | |
| 137 | func TestConvergerCapturesBeforeDeclarationAndRunReads(t *testing.T) { |
| 138 | for _, key := range []string{DeclarationKey("bab", "app"), RunKey("bab", "app")} { |
| 139 | t.Run(key, func(t *testing.T) { |
| 140 | reg, store := convergeFixture(t) |
| 141 | decl := []byte(`{"schedules":[{"name":"tick","every":"1h","path":"/tick"}]}`) |
| 142 | if _, err := store.Put(t.Context(), DeclarationKey("bab", "app"), decl, ""); err != nil { |
| 143 | t.Fatal(err) |
| 144 | } |
| 145 | epoch := &atomic.Int64{} |
| 146 | selector := &rotatingSelector{current: epoch} |
| 147 | c := NewConverger(reg, changeOnRead{store, key, epoch}, "box", nil) |
| 148 | c.HostWrites = hostwrite.Config{Enabled: true, Selector: selector} |
| 149 | if err := c.Pass(t.Context()); err == nil { |
| 150 | t.Fatal("old inputs acquired successor authority") |
| 151 | } |
| 152 | rows, err := reg.TimersFor("a.bab.oncarlos.com") |
| 153 | if err != nil || len(rows) != 0 { |
| 154 | t.Fatalf("stale rows=%v %v", rows, err) |
| 155 | } |
| 156 | if selector.calls != 4 { |
| 157 | t.Fatalf("selection refreshed: %d", selector.calls) |
| 158 | } |
| 159 | }) |
| 160 | } |
| 161 | } |
| 162 | |
| 163 | type replaceAtWork struct { |
| 164 | n, at int |
| 165 | replace func() |
| 166 | } |
| 167 | |
| 168 | func (w *replaceAtWork) With(_ context.Context, _ hostwrite.Identity, fn func() error) error { |
| 169 | w.n++ |
| 170 | if w.n == w.at { |
| 171 | w.replace() |
| 172 | } |
| 173 | return fn() |
| 174 | } |
| 175 | func TestConvergerRegistrationMutationsPreserveReplacement(t *testing.T) { |
| 176 | for _, kind := range []string{"delete", "unit-delete", "status", "run", "upsert"} { |
| 177 | t.Run(kind, func(t *testing.T) { |
| 178 | reg, _ := convergeFixture(t) |
| 179 | host := "a.bab.oncarlos.com" |
| 180 | original := registry.Timer{Host: host, Name: "tick", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/tick", NextAt: 100, DueAt: 100} |
| 181 | if err := reg.UpsertTimer(original); err != nil { |
| 182 | t.Fatal(err) |
| 183 | } |
| 184 | rt, _ := reg.Get(host) |
| 185 | var decl Declaration |
| 186 | if err := json.Unmarshal([]byte(`{"schedules":[{"name":"tick","every":"1h","path":"/tick"}]}`), &decl); err != nil { |
| 187 | t.Fatal(err) |
| 188 | } |
| 189 | fire := "" |
| 190 | at := 2 |
| 191 | switch kind { |
| 192 | case "delete": |
| 193 | decl = Declaration{} |
| 194 | case "unit-delete": |
| 195 | rt.Backing = registry.BackingUnit |
| 196 | decl = Declaration{} |
| 197 | case "status": |
| 198 | decl = Declaration{} |
| 199 | if err := json.Unmarshal([]byte(`{"schedules":[{"name":"tick","cron":"0 0 31 2 *","path":"/tick"}]}`), &decl); err != nil { |
| 200 | t.Fatal(err) |
| 201 | } |
| 202 | at = 3 |
| 203 | case "run": |
| 204 | fire = "tick" |
| 205 | at = 4 |
| 206 | } |
| 207 | var replacement *registry.Timer |
| 208 | w := &replaceAtWork{at: at, replace: func() { |
| 209 | new := original |
| 210 | new.Path = "/successor" |
| 211 | new.NextAt = 999 |
| 212 | new.DueAt = 999 |
| 213 | if err := reg.UpsertTimer(new); err != nil { |
| 214 | t.Fatal(err) |
| 215 | } |
| 216 | replacement, _ = reg.GetTimer(host, "tick") |
| 217 | }} |
| 218 | ctx := context.WithValue(t.Context(), timerWorkKey{}, timerWork{ctx: t.Context(), work: w}) |
| 219 | c := NewConverger(reg, nil, "box", nil) |
| 220 | c.HostWrites.Enabled = true |
| 221 | _, _ = c.convergeHost(ctx, *rt, decl, fire, time.Unix(1000, 0)) |
| 222 | got, err := reg.GetTimer(host, "tick") |
| 223 | if replacement == nil || err != nil || !reflect.DeepEqual(got, replacement) { |
| 224 | t.Fatalf("%s replaced row: got=%+v want=%+v err=%v work=%d", kind, got, replacement, err, w.n) |
| 225 | } |
| 226 | }) |
| 227 | } |
| 228 | } |
| 229 | |
| 230 | func (d pausedDelivery) DeliverClaim(ctx context.Context, rt registry.Route, req *http.Request, admit func(context.Context) (context.Context, func(), error)) (int, error) { |
| 231 | child, finish, err := admit(ctx) |
| 232 | if err != nil { |
| 233 | return 0, err |
| 234 | } |
| 235 | defer finish() |
| 236 | return d.Deliver(child, rt, req) |
| 237 | } |
| 238 | |
amadan