amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package console
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "net/http"
8 "net/http/httptest"
9 "strings"
10 "testing"
11 "time"
12
13 "github.com/carlosframework/platform/internal/hostops"
14 "github.com/carlosframework/platform/internal/mail"
15 "github.com/carlosframework/platform/internal/release"
16)
17
18func TestClaimMutationStaleConsoleCannotMutateRecreatedApp(t *testing.T) {
19 t.Run("bound", func(t *testing.T) { claimStaleIncarnationCase(t, false) })
20 t.Run("unbound", func(t *testing.T) { claimStaleIncarnationCase(t, true) })
21}
22
23func claimStaleIncarnationCase(t *testing.T, unbound bool) {
24 w, _ := newClaimRestoreWorld(t)
25 for _, srv := range []*Server{w.a, w.b} {
26 if err := srv.store.AddMember(crAcctID, crOwner, RoleOwner, time.Now()); err != nil {
27 t.Fatal(err)
28 }
29 }
30 if err := w.a.restoreApp(w.ctx, w.account(t, w.a), crApp, time.Now()); err != nil {
31 t.Fatal(err)
32 }
33 w.materialize(t, w.b)
34 // B retains its queued callbacks while A purges and recreates the name.
35 w.b.cfg.AppsDomain = "oncarlos.com"
36 rec := &claimRecorder{}
37 priorTransport := mailPushClient.Transport
38 mailPushClient.Transport = claimOKTransport{rec: rec}
39 t.Cleanup(func() { mailPushClient.Transport = priorTransport })
40 if err := w.b.store.SetMailEventHook(w.ctx, crSqid, crApp, "/mail/events", crOwner, time.Now()); err != nil {
41 t.Fatal(err)
42 }
43 enqueue := func(notice string) {
44 t.Helper()
45 if err := w.b.store.EnqueueMailEventDelivery(w.ctx, crSqid, crApp, notice, []byte(`{}`), time.Now().Add(-time.Minute)); err != nil {
46 t.Fatal(err)
47 }
48 }
49 enqueue("before-recreation")
50 if err := w.b.mailPushPass(w.ctx, time.Now); err != nil {
51 t.Fatal(err)
52 }
53 if events := rec.since(0); len(events) != 1 {
54 t.Fatalf("valid worker did not send before recreation: %v", events)
55 }
56 enqueue("delayed-before-recreation")
57 oldHooks, err := w.b.store.activeMailEventHooks(w.ctx)
58 if err != nil || len(oldHooks) != 1 {
59 t.Fatalf("old hooks: %v %v", oldHooks, err)
60 }
61 oldHook := oldHooks[0]
62 oldHook.claimAppID = crAppID
63 oldRows, err := w.b.store.dueMailEventDeliveriesFor(w.ctx, crSqid, crApp, time.Now(), mailPushPerApp)
64 if err != nil || len(oldRows) != 1 {
65 t.Fatalf("old rows: %v %v", oldRows, err)
66 }
67
68 // B goes offline holding A's row and credentials. Stage the durable result
69 // of A's completed purge; the independent purge suite drives its chores.
70 staleToken := mintUserToken(t, w.b.store, crOwner, time.Now())
71 stale := httptest.NewServer(w.b)
72 defer stale.Close()
73 body := `{"set":{"REV5":{"kind":"env","value":"stale"}}}`
74 for _, token := range []string{staleToken} {
75 status, _, out := rawReq(t, http.MethodPut, stale.URL+"/api/cli/apps/bab/kass/config", token, body)
76 if status != http.StatusOK {
77 t.Fatalf("old credential was not a valid mutation fixture: %d %s", status, out)
78 }
79 }
80 keys, err := w.bucket.List(w.ctx, "apps/"+crSqid+"/"+crApp+"/")
81 if err != nil {
82 t.Fatal(err)
83 }
84 for _, key := range keys {
85 if !strings.HasSuffix(key, "/admission.json") {
86 if err := w.bucket.Delete(w.ctx, key); err != nil {
87 t.Fatal(err)
88 }
89 }
90 }
91 a := w.admission(t).Value
92 a.State = "closed"
93 a.Op = nil
94 a.Registrations = map[string]hostops.Registration{}
95 a.Effects = map[string]hostops.Effect{}
96 a.LastPurge = &hostops.PurgeMark{ID: "finished-purge", AppID: crAppID}
97 seedClaimAdmission(t, w.bucket, crSqid, crApp, a)
98 err = w.a.store.withClaimSQL(gateCtx(t), []claimSQLScope{{AccountID: crAcctID, App: crApp, AppID: crAppID, Mode: claimModeEffect}}, allowAllClaimSQL, func(st *Store) error {
99 _, err := st.db.Exec(`DELETE FROM apps WHERE account_id=? AND app=?`, crAcctID, crApp)
100 return err
101 })
102 if err != nil {
103 t.Fatal(err)
104 }
105 // The production creation path reserves and binds B's new incarnation.
106 created, err := w.a.createClaimAppRow(w.ctx, w.account(t, w.a), crApp, func(st *Store) (bool, error) { return st.CreateAppIn(crAcctID, crApp, crOwner, time.Now()) })
107 if err != nil || !created {
108 t.Fatalf("recreate: %v %v", created, err)
109 }
110 current, err := hostops.ReadCanonicalAdmission(w.ctx, w.bucket, crSqid, crApp)
111 if err != nil {
112 t.Fatal(err)
113 }
114 if current.Value.AppID == crAppID {
115 t.Fatal("recreation reused incarnation")
116 }
117 if local, err := w.b.store.ClaimAppID(crAcctID, crApp); err != nil || local != crAppID {
118 t.Fatalf("stale row %q %v", local, err)
119 }
120
121 if unbound {
122 err = w.b.store.withClaimSQL(gateCtx(t), []claimSQLScope{{AccountID: crAcctID, App: crApp, AppID: crAppID, Mode: claimModeEffect}}, allowAllClaimSQL, func(st *Store) error {
123 _, err := st.db.Exec(`UPDATE apps SET app_id='' WHERE account_id=? AND app=?`, crAcctID, crApp)
124 return err
125 })
126 if err != nil {
127 t.Fatal(err)
128 }
129 if err = w.b.bindExistingClaimAppAdmission(w.ctx, w.account(t, w.b), crApp); !errors.Is(err, hostops.ErrPending) {
130 t.Fatalf("unbound stale row adopted successor: %v", err)
131 }
132 if err = w.b.SweepClaimAppIdentities(w.ctx); err != nil {
133 t.Fatal(err)
134 }
135 if id, err := w.b.store.ClaimAppID(crAcctID, crApp); id != "" || !errors.Is(err, hostops.ErrIdentity) {
136 t.Fatalf("sweep bound untrusted identity: %q %v", id, err)
137 }
138 }
139 // A queued in-memory turn cannot bind to the same console's new row.
140 w.a.cfg.AppsDomain = "oncarlos.com"
141 if err := w.a.store.SetMailEventHook(w.ctx, crSqid, crApp, "/mail/events", crOwner, time.Now()); err != nil {
142 t.Fatal(err)
143 }
144 signingKey, err := w.a.store.MailEventsSigningKey()
145 if err != nil {
146 t.Fatal(err)
147 }
148 delayedMark := rec.mark()
149 w.a.deliverAppMailEvents(w.ctx, oldHook, oldRows, signingKey, time.Now)
150 if events := rec.since(delayedMark); len(events) != 0 {
151 t.Fatalf("delayed turn sent to successor: %v", events)
152 }
153 current, err = hostops.ReadCanonicalAdmission(w.ctx, w.bucket, crSqid, crApp)
154 if err != nil {
155 t.Fatal(err)
156 }
157 mark := rec.mark()
158 if err := w.b.mailPushPass(w.ctx, time.Now); err != nil {
159 t.Fatal(err)
160 }
161 if events := rec.since(mark); len(events) != 0 {
162 t.Fatalf("stale worker sent after recreation: %v", events)
163 }
164 pending, err := w.b.store.dueMailEventDeliveriesFor(w.ctx, crSqid, crApp, time.Now(), mailPushPerApp)
165 if err != nil || len(pending) != 1 {
166 t.Fatalf("stale worker settled old queue: %v %v", pending, err)
167 }
168 for _, token := range []string{staleToken} {
169 status, _, out := rawReq(t, http.MethodPut, stale.URL+"/api/cli/apps/bab/kass/config", token, body)
170 if status != http.StatusBadGateway {
171 t.Fatalf("stale console mutated successor: %d %s", status, out)
172 }
173 after, err := hostops.ReadCanonicalAdmission(w.ctx, w.bucket, crSqid, crApp)
174 if err != nil {
175 t.Fatal(err)
176 }
177 if after.ETag != current.ETag {
178 t.Fatal("stale request acquired successor admission")
179 }
180 keys, err = w.bucket.List(w.ctx, "apps/"+crSqid+"/"+crApp+"/config/")
181 if err != nil || len(keys) != 0 {
182 t.Fatalf("stale config effect: %v %v", keys, err)
183 }
184 }
185 called := false
186 err = release.WithAppMutation(w.ctx, w.b.cfg.Releases, crSqid, crApp, "stale", func(context.Context) error { called = true; return nil })
187 if !errors.Is(err, hostops.ErrIdentity) || called {
188 t.Fatalf("stale runner: called=%v err=%v", called, err)
189 }
190 _, err = mail.Declare(w.ctx, w.b.claimMailStore(), crSqid, crApp, "EMAIL", "sender@example.com", crOwner, time.Now())
191 if !errors.Is(err, hostops.ErrIdentity) {
192 t.Fatalf("stale mail admitted: %v", err)
193 }
194 after, err := hostops.ReadCanonicalAdmission(w.ctx, w.bucket, crSqid, crApp)
195 if err != nil || after.ETag != current.ETag {
196 t.Fatalf("stale mail changed successor admission: %v", err)
197 }
198 gate, _, err := w.bucket.Get(w.ctx, hostops.MailGateKey)
199 if err != nil {
200 t.Fatal(err)
201 }
202 var held struct {
203 State string `json:"state"`
204 Members []json.RawMessage `json:"members"`
205 }
206 if err = json.Unmarshal(gate, &held); err != nil || held.State != "free" || len(held.Members) != 0 {
207 t.Fatalf("stale mail left gate membership: %s %v", gate, err)
208 }
209 fresh := httptest.NewServer(w.a)
210 defer fresh.Close()
211 token := mintUserToken(t, w.a.store, crOwner, time.Now())
212 status, _, out := rawReq(t, http.MethodPut, fresh.URL+"/api/cli/apps/bab/kass/config", token, body)
213 if status != http.StatusOK {
214 t.Fatalf("current console refused: %d %s", status, out)
215 }
216}
217
218func TestClaimAdmissionSweepContinuesPastStaleLocalIdentity(t *testing.T) {
219 w, _ := newClaimRestoreWorld(t)
220 a := w.admission(t).Value
221 a.AppID = "successor"
222 a.LastPurge = &hostops.PurgeMark{ID: "finished", AppID: crAppID}
223 seedClaimAdmission(t, w.bucket, crSqid, crApp, a)
224 op := hostops.AppOperation{ID: "joined", Kind: hostops.OpAppMutation, AppID: "other-id", Intent: map[string]string{"action": "test"}}
225 pending := hostops.AppAdmission{V: 1, AppID: "other-id", Revision: 1, Epoch: 1, State: "closed", Op: &op, Registrations: map[string]hostops.Registration{}, Effects: map[string]hostops.Effect{"joined": {ID: "joined", OperationID: "joined", AppID: "other-id", State: "complete", Resource: hostops.OpAppMutation, Action: "test", Owner: w.b.claimOwner}}}
226 seedClaimAdmission(t, w.bucket, crSqid, "other", pending)
227 if err := w.b.SweepClaimAdmissions(w.ctx); err != nil {
228 t.Fatal(err)
229 }
230 other, err := hostops.ReadAdmission(w.ctx, w.bucket, crSqid, "other", "other-id")
231 if err != nil {
232 t.Fatal(err)
233 }
234 if other.Value.Op != nil || other.Value.State != "open" {
235 t.Fatal("stale local row blocked another app's recovery")
236 }
237 if id, err := w.b.store.ClaimAppID(crAcctID, crApp); err != nil || id != crAppID {
238 t.Fatalf("sweep rebound stale identity: %q %v", id, err)
239 }
240}
241
242func TestClaimAdmissionSweepContinuesPastMalformedIdentity(t *testing.T) {
243 w, _ := newClaimRestoreWorld(t)
244 a := hostops.AppAdmission{V: 1, AppID: "other-id", Revision: 1, Epoch: 1, State: "closed", Op: &hostops.AppOperation{ID: "joined", Kind: hostops.OpAppMutation, AppID: "other-id", Intent: map[string]string{"action": "test"}}, Registrations: map[string]hostops.Registration{}, Effects: map[string]hostops.Effect{"joined": {ID: "joined", OperationID: "joined", AppID: "other-id", State: "complete", Owner: w.b.claimOwner}}}
245 seedClaimAdmission(t, w.bucket, crSqid, "other", a)
246 key := "apps/" + crSqid + "/" + crApp + "/admission.json"
247 _, tag, err := w.bucket.Get(w.ctx, key)
248 if err != nil {
249 t.Fatal(err)
250 }
251 if _, err = w.bucket.Put(w.ctx, key, []byte(`{broken`), tag); err != nil {
252 t.Fatal(err)
253 }
254 if err = w.b.SweepClaimAdmissions(w.ctx); err == nil {
255 t.Fatal("malformed identity was not reported")
256 }
257 other, err := hostops.ReadAdmission(w.ctx, w.bucket, crSqid, "other", "other-id")
258 if err != nil {
259 t.Fatal(err)
260 }
261 if other.Value.Op != nil || other.Value.State != "open" {
262 t.Fatal("identity error stopped independent recovery")
263 }
264}
265
266func TestClaimMailPreflightsAllLocalIdentitiesBeforeAnyMember(t *testing.T) {
267 w, _ := newClaimRestoreWorld(t)
268 const other = "aaa"
269 for _, srv := range []*Server{w.a, w.b} {
270 if _, err := srv.createClaimAppRow(w.ctx, w.account(t, srv), other, func(st *Store) (bool, error) { return st.CreateAppIn(crAcctID, other, crOwner, time.Now()) }); err != nil {
271 t.Fatal(err)
272 }
273 }
274 a := w.admission(t).Value
275 a.AppID = "successor"
276 a.LastPurge = &hostops.PurgeMark{ID: "finished", AppID: crAppID}
277 seedClaimAdmission(t, w.bucket, crSqid, crApp, a)
278 before, err := hostops.ReadCanonicalAdmission(w.ctx, w.bucket, crSqid, other)
279 if err != nil {
280 t.Fatal(err)
281 }
282 _, err = mail.PutDomain(w.ctx, w.b.claimMailStore(), crSqid, &mail.DomainRecord{Schema: 1, Domain: "shared.example.com", Apps: []string{other, crApp}}, "")
283 if !errors.Is(err, hostops.ErrIdentity) {
284 t.Fatalf("stale shared-domain member: %v", err)
285 }
286 after, err := hostops.ReadCanonicalAdmission(w.ctx, w.bucket, crSqid, other)
287 if err != nil || after.ETag != before.ETag {
288 t.Fatalf("earlier member was admitted before complete identity preflight: %v", err)
289 }
290 gate, _, err := w.bucket.Get(w.ctx, hostops.MailGateKey)
291 if err != nil {
292 t.Fatal(err)
293 }
294 var held struct {
295 State string `json:"state"`
296 Members []json.RawMessage `json:"members"`
297 }
298 if err = json.Unmarshal(gate, &held); err != nil || held.State != "free" || len(held.Members) != 0 {
299 t.Fatalf("refusal recorded members: %s %v", gate, err)
300 }
301}
302