carlos / platform Public

Download this file

1	package host
2
3	import (
4	"context"
5	"errors"
6	"io"
7	"net/http"
8	"net/http/httptest"
9	"os"
10	"path/filepath"
11	"sync/atomic"
12	"testing"
13	"time"
14
15	"github.com/carlosframework/platform/internal/activator"
16	"github.com/carlosframework/platform/internal/edge"
17	"github.com/carlosframework/platform/internal/instancectl"
18	"github.com/carlosframework/platform/internal/layout"
19	"github.com/carlosframework/platform/internal/registry"
20	)
21
22	type startupBackend struct {
23	fakeWakeBackend
24	recovered *atomic.Bool
25	started *atomic.Bool
26	attempts atomic.Int32
27	gate func() error
28	}
29
30	func (b *startupBackend) Acquire(ctx context.Context, rt registry.Route) (activator.Lease, error) {
31	b.attempts.Add(1)
32	if !b.recovered.Load() || !b.started.Load() {
33	return nil, errors.New("wake preceded recovery or Start")
34	}
35	if b.gate != nil {
36	if err := b.gate(); err != nil {
37	return nil, err
38	}
39	}
40	return b.fakeWakeBackend.Acquire(ctx, rt)
41	}
42
43	type observingActivator struct {
44	*activator.Service
45	entered chan struct{}
46	}
47
48	func (a observingActivator) ServeHTTP(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler) bool {
49	a.entered <- struct{}{}
50	return a.Service.ServeHTTP(w, r, rt, proxy)
51	}
52
53	func TestStartupServesWhileSweepAndRecoveryWait(t *testing.T) {
54	t.Setenv("CARLOS_STEERING_ZONE_ID", "Z1234567890")
55	t.Setenv("PLATFORM_APPS_DOMAIN", "apps.example.com")
56	for _, fail := range []bool{false, true} {
57	t.Run(map[bool]string{false: "slow sweep and recovery", true: "unreachable instance control"}[fail], func(t *testing.T) {
58	dd, err := os.MkdirTemp(os.TempDir(), "startup-")
59	if err != nil {
60	t.Fatal(err)
61	}
62	t.Cleanup(func() { os.RemoveAll(dd) })
63	reg, err := registry.Open(filepath.Join(dd, "registry.db"))
64	if err != nil {
65	t.Fatal(err)
66	}
67	defer reg.Close()
68	inst := &fakeInstance{socket: filepath.Join(dd, "app.sock"), handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { io.WriteString(w, "awake") })}
69	rt := registry.Route{Host: "shop.example.com", Kind: "instance", Hibernate: true, Status: "ready", Socket: inst.socket, DBPath: filepath.Join(dd, "app.db")}
70	if err := reg.Upsert(rt); err != nil {
71	t.Fatal(err)
72	}
73	console := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) }))
74	defer console.Close()
75	if err := reg.Upsert(registry.Route{Host: "console.example.com", Addr: console.Listener.Addr().String()}); err != nil {
76	t.Fatal(err)
77	}
78	var recovered, startedService atomic.Bool
79	backend := &startupBackend{fakeWakeBackend: fakeWakeBackend{inst: inst}, recovered: &recovered, started: &startedService}
80	control := instancectl.NewClient(filepath.Join(dd, "unreachable.sock"))
81	if fail {
82	backend.gate = nestedGate(func() error { return nil }, func(string) (bool, error) { return true, nil }, control.Ready)
83	}
84	svc := activator.New(activator.Config{Bucket: "unused", Idle: time.Hour}, backend, reg)
85	ready := make(chan struct{})
86	svc.Startup = ready
87	ctx, cancel := context.WithCancel(t.Context())
88	defer cancel()
89	defer func() { svc.Shutdown(context.Background()) }()
90	sweepEntered, recoverEntered := make(chan struct{}), make(chan struct{})
91	sweepRelease, recoverRelease := make(chan struct{}), make(chan struct{})
92	done := activatorStart{
93	Sweep: func(ctx context.Context) error {
94	close(sweepEntered)
95	select {
96	case <-sweepRelease:
97	case <-ctx.Done():
98	return ctx.Err()
99	}
100	if fail {
101	_, err := control.Sweep(ctx)
102	return err
103	}
104	return nil
105	},
106	Recover: func(ctx context.Context, swept bool) *layout.Pass {
107	if swept == fail {
108	t.Error("recovery received the wrong sweep result")
109	}
110	close(recoverEntered)
111	select {
112	case <-recoverRelease:
113	case <-ctx.Done():
114	return nil
115	}
116	recovered.Store(true)
117	return nil
118	},
119	Publish: func(*layout.Pass) {}, Start: func() { svc.Start(); startedService.Store(true) }, Ready: func() { close(ready) },
120	Routes: func() ([]registry.Route, error) { return nil, nil },
121	}.Run(ctx)
122	defer func() { cancel(); <-done }()
123	<-sweepEntered
124	entered := make(chan struct{}, 1)
125	e, err := edge.New(reg, observingActivator{svc, entered})
126	if err != nil {
127	t.Fatal(err)
128	}
129	server := httptest.NewServer(e)
130	defer server.Close()
131	client := &http.Client{Timeout: time.Second}
132	get := func(host string) (int, error) {
133	r, _ := http.NewRequest(http.MethodGet, server.URL+"/", nil)
134	r.Host = host
135	resp, err := client.Do(r)
136	if err != nil {
137	return 0, err
138	}
139	io.Copy(io.Discard, resp.Body)
140	resp.Body.Close()
141	return resp.StatusCode, nil
142	}
143	for host, want := range map[string]int{"127.0.0.1": http.StatusNotFound, "console.example.com": http.StatusNoContent, "edge.apps.example.com": http.StatusOK} {
144	began := time.Now()
145	if got, err := get(host); err != nil || got != want {
146	t.Fatalf("%s during sweep: %d, %v", host, got, err)
147	}
148	t.Logf("%s answered in %s while sweep was blocked", host, time.Since(began))
149	}
150	answer := make(chan int, 1)
151	go func() { status, _ := get(rt.Host); answer <- status }()
152	select {
153	case <-entered:
154	case <-time.After(time.Second):
155	t.Fatal("request never reached the activator")
156	}
157	assertHeld := func() {
158	t.Helper()
159	select {
160	case status := <-answer:
161	t.Fatalf("request answered before startup: %d", status)
162	case <-time.After(25 * time.Millisecond):
163	}
164	if backend.attempts.Load() != 0 {
165	t.Fatal("wake attempted before sweep and recovery finished")
166	}
167	}
168	assertHeld()
169	close(sweepRelease)
170	<-recoverEntered
171	assertHeld()
172	if got, err := get("console.example.com"); err != nil || got != http.StatusNoContent {
173	t.Fatalf("console during recovery: %d, %v", got, err)
174	}
175	close(recoverRelease)
176	want := http.StatusOK
177	if fail {
178	want = http.StatusServiceUnavailable
179	}
180	select {
181	case got := <-answer:
182	if got != want {
183	t.Fatalf("instance after Start: %d, want %d", got, want)
184	}
185	case <-time.After(time.Second):
186	t.Fatal("request did not finish after Start")
187	}
188	if backend.attempts.Load() != 1 || inst.running() == fail {
189	t.Fatal("unexpected wake outcome")
190	}
191	})
192	}
193	}
194
195	func TestStartupCancellationAndBoundedJoin(t *testing.T) {
196	t.Parallel()
197	ctx, cancel := context.WithCancel(context.Background())
198	entered := make(chan struct{})
199	var started atomic.Bool
200	done := activatorStart{
201	Sweep: func(ctx context.Context) error { close(entered); <-ctx.Done(); return ctx.Err() },
202	Recover: func(context.Context, bool) *layout.Pass { t.Error("recovery after cancellation"); return nil },
203	Start: func() { started.Store(true) },
204	}.Run(ctx)
205	<-entered
206	cancel()
207	join, stop := context.WithTimeout(t.Context(), time.Second)
208	defer stop()
209	if !joinAgentStartup(join, done) || started.Load() {
210	t.Fatal("cancelled startup was not joined, or started")
211	}
212	blocked := make(chan struct{})
213	expired, expire := context.WithTimeout(t.Context(), 20*time.Millisecond)
214	defer expire()
215	if joinAgentStartup(expired, blocked) {
216	t.Fatal("join ignored its bound")
217	}
218	}
219
220	func TestStartupJoinsKeyPassOnAgentCancellation(t *testing.T) {
221	ctx, cancel := context.WithCancel(t.Context())
222	defer cancel()
223	entered := make(chan struct{})
224	rt := registry.Route{Host: "shop.example.com", Kind: "instance", Socket: "/unused.sock", Status: "ready"}
225	done := activatorStart{
226	Sweep: func(context.Context) error { return nil },
227	Recover: func(context.Context, bool) *layout.Pass { return nil },
228	Publish: func(*layout.Pass) {}, Start: func() {},
229	Routes: func() ([]registry.Route, error) { return []registry.Route{rt}, nil },
230	Route: routeOf(rt), Slot: openSlot,
231	Keys: keyBackfillFunc(func(ctx context.Context, _ registry.Route) (activator.InstanceKeyOutcome, error) {
232	close(entered)
233	<-ctx.Done()
234	return activator.InstanceKeyFailed, ctx.Err()
235	}),
236	}.Run(ctx)
237	select {
238	case <-entered:
239	case <-time.After(time.Second):
240	t.Fatal("key pass did not start")
241	}
242	cancel()
243	join, stop := context.WithTimeout(t.Context(), time.Second)
244	defer stop()
245	if !joinAgentStartup(join, done) {
246	t.Fatal("key pass outlived agent cancellation")
247	}
248	}
249
250	func TestStartupCancellationDuringRecoveryPreservesAgentLifetime(t *testing.T) {
251	agentCtx, cancelAgent := context.WithCancel(t.Context())
252	defer cancelAgent()
253	startupCtx, cancelStartup := context.WithCancel(agentCtx)
254	defer cancelStartup()
255	recovering := make(chan struct{})
256	svc := activator.New(activator.Config{}, &fakeWakeBackend{}, nil)
257	ready := make(chan struct{})
258	svc.Startup = ready
259	done := activatorStart{
260	Configure: func(context.Context) {
261	if agentCtx.Err() != nil {
262	t.Error("agent lifetime ended before configuration")
263	}
264	},
265	Sweep: func(context.Context) error { return nil },
266	Recover: func(ctx context.Context, _ bool) *layout.Pass {
267	close(recovering)
268	<-ctx.Done()
269	return nil
270	},
271	Publish: func(*layout.Pass) {},
272	Start: func() { t.Error("Start after recovery cancellation") },
273	Ready: func() { close(ready) },
274	}.Run(startupCtx)
275	<-recovering
276	cancelStartup()
277	join, stop := context.WithTimeout(t.Context(), time.Second)
278	defer stop()
279	if !joinAgentStartup(join, done) {
280	t.Fatal("recovery did not join")
281	}
282	svc.Shutdown(join)
283	if agentCtx.Err() != nil {
284	t.Fatal("startup cancellation stopped rollout's agent lifetime before children drained")
285	}
286	select {
287	case <-ready:
288	t.Fatal("cancelled recovery released admission")
289	default:
290	}
291	}
292