carlos / platform Public

Download this file

1	package hostops
2
3	import (
4	"context"
5	"encoding/json"
6	"errors"
7	"github.com/carlosframework/platform/internal/release"
8	"testing"
9	"time"
10	)
11
12	func TestIngestSharedDrainAndExpiredIssuedEffect(t *testing.T) {
13	w, raw, now := testWriter(t)
14	ctx := t.Context()
15	seedAdmission(t, raw, "bab", "demo", openAdmission("app"))
16	var handles []*IngestHandle
17	var settles []func(string, error) error
18	for _, id := range []string{"ingest-a", "ingest-b"} {
19	h, err := w.RegisterIngest(ctx, "bab", "demo", "app", id, testOwner)
20	if err != nil {
21	t.Fatal(err)
22	}
23	handles = append(handles, h)
24	pctx, settle, err := h.start.Runner(w.now).Issue(ctx, Effect{ID: id, Resource: "app-sql:" + id, Action: sqlIngestAction, Inputs: map[string]string{"sql_database": "local"}})
25	if err != nil {
26	t.Fatal(err)
27	}
28	settles = append(settles, settle)
29	p, err := PrepareSQLPermit(pctx, raw, "bab", "demo", "app")
30	if err != nil {
31	t.Fatal(err)
32	}
33	if err = RequireAppMutation(pctx, raw, "bab", "demo", "app"); err == nil {
34	t.Fatal("SQL permit granted generic app write")
35	}
36	if generic, err := LivePermits(pctx, raw); err != nil || len(generic) != 0 {
37	t.Fatalf("SQL-only permit leaked to backend: %v %v", generic, err)
38	}
39	if err = p.Check(ctx); err != nil {
40	t.Fatal(err)
41	}
42	}
43	op, err := w.BeginExclusive(ctx, "bab", "demo", "app", AppOperation{ID: "delete", Kind: OpDeleting, AppID: "app"}, testOwner)
44	if err != nil {
45	t.Fatal(err)
46	}
47	later := now().Add(2 * RegistrationTTL)
48	if err = op.Drain(ctx, later, nil); !errors.Is(err, ErrPending) {
49	t.Fatalf("expired issued effects drained: %v", err)
50	}
51	if err = settles[0]("sql-joined", nil); err != nil {
52	t.Fatal(err)
53	}
54	if err = op.Drain(ctx, later, nil); !errors.Is(err, ErrPending) {
55	t.Fatalf("drain ignored second ingest: %v", err)
56	}
57	if err = settles[1]("sql-joined", nil); err != nil {
58	t.Fatal(err)
59	}
60	if err = op.Drain(ctx, later, nil); err != nil {
61	t.Fatal(err)
62	}
63	if _, err = w.RegisterIngest(ctx, "bab", "demo", "app", "late", testOwner); !errors.Is(err, ErrPending) {
64	t.Fatalf("closed admission: %v", err)
65	}
66	_ = handles
67	}
68
69	type sqlProof struct {
70	fail bool
71	seen SQLReceiptScope
72	}
73
74	func (p *sqlProof) FenceSQL(_ context.Context, s SQLReceiptScope) error {
75	p.seen = s
76	if p.fail {
77	return errors.New("database unavailable")
78	}
79	return nil
80	}
81	func TestIngestRecoveryRequiresSQLFence(t *testing.T) {
82	w, raw, _ := testWriter(t)
83	ctx := t.Context()
84	seedAdmission(t, raw, "bab", "demo", openAdmission("app"))
85	h, err := w.RegisterIngest(ctx, "bab", "demo", "app", "ingest", testOwner)
86	if err != nil {
87	t.Fatal(err)
88	}
89	_, settle, err := h.start.Runner(w.now).Issue(ctx, Effect{ID: "ingest", Resource: "app-sql:ingest", Action: sqlIngestAction, Inputs: map[string]string{"sql_database": "original"}})
90	if err != nil {
91	t.Fatal(err)
92	}
93	_ = settle("", errors.New("console vanished"))
94	proof := &sqlProof{fail: true}
95	if err = w.RecoverIngest(ctx, "bab", "demo", "app", "ingest", proof); err == nil {
96	t.Fatal("missing database settled")
97	}
98	a, _ := ReadAdmission(ctx, raw, "bab", "demo", "app")
99	if a.Value.Effects["ingest"].State != "uncertain" {
100	t.Fatal("uncertainty cleared")
101	}
102	proof.fail = false
103	if err = w.RecoverIngest(ctx, "bab", "demo", "app", "ingest", proof); err != nil {
104	t.Fatal(err)
105	}
106	if proof.seen.Database != "original" || proof.seen.Owner != testOwner {
107	t.Fatal("recovery scope changed")
108	}
109	a, _ = ReadAdmission(ctx, raw, "bab", "demo", "app")
110	if len(a.Value.Registrations) != 0 || a.Value.Effects["ingest"].ResultReference != "sql-fenced" {
111	t.Fatal("recovery did not drain")
112	}
113	}
114	func TestPreparedSQLCancellationDeadlineAndRevocation(t *testing.T) {
115	w, raw, _ := testWriter(t)
116	ctx, cancel := context.WithCancel(t.Context())
117	defer cancel()
118	seedAdmission(t, raw, "bab", "demo", openAdmission("app"))
119	h, err := w.RegisterIngest(ctx, "bab", "demo", "app", "ingest", testOwner)
120	if err != nil {
121	t.Fatal(err)
122	}
123	pctx, settle, err := h.start.Runner(w.now).Issue(ctx, Effect{ID: "ingest", Resource: "app-sql:ingest", Action: sqlIngestAction, Inputs: map[string]string{"sql_database": "db"}})
124	if err != nil {
125	t.Fatal(err)
126	}
127	p, err := PrepareSQLPermit(pctx, raw, "bab", "demo", "app")
128	if err != nil {
129	t.Fatal(err)
130	}
131	if _, err = PrepareSQLPermit(pctx, raw, "bab", "other", "app"); err == nil {
132	t.Fatal("cross-app accepted")
133	}
134	p.deadline = time.Now().Add(-time.Second)
135	if !errors.Is(p.Check(ctx), context.DeadlineExceeded) {
136	t.Fatal("deadline ignored")
137	}
138	p.deadline = time.Now().Add(time.Second)
139	cancel()
140	if !errors.Is(p.Check(context.Background()), context.Canceled) {
141	t.Fatal("request cancellation ignored")
142	}
143	_ = settle("sql-joined", nil)
144	if p.Check(context.Background()) == nil {
145	t.Fatal("settled permit survived")
146	}
147	}
148
149	func TestPreparedSQLChildBlocksRestoreRecovery(t *testing.T) {
150	w, raw, _ := testWriter(t)
151	ctx := t.Context()
152	seedAdmission(t, raw, "bab", "demo", openAdmission("app"))
153	h, err := w.BeginExclusive(ctx, "bab", "demo", "app", AppOperation{ID: "restore", Kind: OpAppMutation, AppID: "app", Intent: map[string]string{"action": "restore"}}, testOwner)
154	if err != nil {
155	t.Fatal(err)
156	}
157	if err = h.PrepareRestore(ctx, "old-marker", 1); err != nil {
158	t.Fatal(err)
159	}
160	pctx, _, err := h.Runner(w.now).Issue(ctx, Effect{ID: "restore", Resource: OpAppMutation, Action: "restore"})
161	if err != nil {
162	t.Fatal(err)
163	}
164	p, err := PrepareSQLPermit(pctx, raw, "bab", "demo", "app")
165	if err != nil {
166	t.Fatal(err)
167	}
168	if err = p.BindTransaction("original-db"); err != nil {
169	t.Fatal(err)
170	}
171	child := p.Receipt()
172	if child.Effect == "restore" || child.Operation != "restore" {
173	t.Fatalf("not a SQL child: %+v", child)
174	}
175	if err = w.RecoverMutation(ctx, "bab", "demo", "app", "restore"); !errors.Is(err, ErrPending) {
176	t.Fatalf("restore escaped issued SQL child: %v", err)
177	}
178	a, _ := ReadAdmission(ctx, raw, "bab", "demo", "app")
179	if a.Value.Op == nil || a.Value.Effects[child.Effect].State != "issued" {
180	t.Fatal("restore released child obligation")
181	}
182	proof := &sqlProof{}
183	if err = w.RecoverSQLTransaction(ctx, "bab", "demo", "app", child.Effect, proof); err != nil {
184	t.Fatal(err)
185	}
186	if err = w.RecoverMutation(ctx, "bab", "demo", "app", "restore"); err != nil {
187	t.Fatal(err)
188	}
189	a, _ = ReadAdmission(ctx, raw, "bab", "demo", "app")
190	if a.Value.Op != nil {
191	t.Fatal("restore did not finish after SQL proof")
192	}
193	}
194
195	func TestIngestPurgeJournalsBothHostlessDrainObligations(t *testing.T) {
196	w, raw, clock := testWriter(t)
197	ctx := t.Context()
198	seedAdmission(t, raw, "bab", "demo", openAdmission("app"))
199	var settles []func(string, error) error
200	for _, id := range []string{"one", "two"} {
201	h, err := w.RegisterIngest(ctx, "bab", "demo", "app", id, testOwner)
202	if err != nil {
203	t.Fatal(err)
204	}
205	_, settle, err := h.start.Runner(w.now).Issue(ctx, Effect{ID: id, Resource: "app-sql:" + id, Action: sqlIngestAction, Inputs: map[string]string{"sql_database": "db"}})
206	if err != nil {
207	t.Fatal(err)
208	}
209	settles = append(settles, settle)
210	}
211	key := release.AppTombstoneKey("bab", "demo")
212	tag, err := raw.Put(ctx, key, []byte(`{}`), "")
213	if err != nil {
214	t.Fatal(err)
215	}
216	journal := PurgeJournal{V: 1, Purge: Purge{ID: "purge", Phase: "reserved"}, AppID: "app", Account: "bab", App: "demo", Hosts: map[string]PurgeHost{}, Tombstone: PurgeObject{Key: key, ETag: tag, Body: json.RawMessage(`{}`)}}
217	if _, err = w.ReservePurge(ctx, journal, testOwner, clock()); err != nil {
218	t.Fatal(err)
219	}
220	if _, err = w.BeginPurge(ctx, "bab", "demo", "app", "purge", testOwner); err != nil {
221	t.Fatal(err)
222	}
223	later := clock().Add(2 * RegistrationTTL)
224	if err = w.DrainPurgeRegistrations(ctx, "bab", "demo", "app", "purge", later, nil); !errors.Is(err, ErrPending) {
225	t.Fatalf("expired SQL obligations drained: %v", err)
226	}
227	if err = settles[0]("sql-joined", nil); err != nil {
228	t.Fatal(err)
229	}
230	if err = w.DrainPurgeRegistrations(ctx, "bab", "demo", "app", "purge", later, nil); !errors.Is(err, ErrPending) {
231	t.Fatalf("second SQL obligation ignored: %v", err)
232	}
233	if err = settles[1]("sql-joined", nil); err != nil {
234	t.Fatal(err)
235	}
236	if err = w.DrainPurgeRegistrations(ctx, "bab", "demo", "app", "purge", later, nil); err != nil {
237	t.Fatal(err)
238	}
239	journal, _, err = ReadPurgeJournal(ctx, raw, "bab", "demo", "purge", "app")
240	if err != nil {
241	t.Fatal(err)
242	}
243	if len(journal.IngestRegistrations) != 2 || len(journal.Hosts) != 0 {
244	t.Fatalf("hostless evidence missing or invented host: %+v", journal)
245	}
246	}
247