carlos / platform Public

Download this file

1	package console
2
3	import (
4	"bytes"
5	"context"
6	"encoding/json"
7	"errors"
8	"fmt"
9	"io"
10	"net/http"
11	"os"
12	"sync/atomic"
13	"testing"
14	"time"
15
16	"github.com/carlosframework/platform/internal/dbdoor"
17	"github.com/carlosframework/platform/internal/lease"
18	)
19
20	type doorRangeHook struct {
21	doorOpenHook
22	openRange func(context.Context, string, string, string, int64, int64) (io.ReadCloser, error)
23	}
24
25	func (v doorRangeHook) OpenVersionRange(ctx context.Context, b, k, id string, offset, length int64) (io.ReadCloser, error) {
26	return v.openRange(ctx, b, k, id, offset, length)
27	}
28
29	type doorCountedReader struct {
30	io.ReadCloser
31	ctx context.Context
32	read *atomic.Int64
33	delay time.Duration
34	}
35
36	func (r *doorCountedReader) Read(p []byte) (int, error) {
37	if r.delay > 0 {
38	timer := time.NewTimer(r.delay)
39	defer timer.Stop()
40	select {
41	case <-r.ctx.Done():
42	return 0, r.ctx.Err()
43	case <-timer.C:
44	}
45	if len(p) > 4096 {
46	p = p[:4096]
47	}
48	}
49	n, err := r.ReadCloser.Read(p)
50	r.read.Add(int64(n))
51	return n, err
52	}
53
54	func TestDoorPlanDeepRangeReadsOnlyRequestedBytes(t *testing.T) {
55	for _, ranged := range []bool{false, true} {
56	t.Run(fmt.Sprint(ranged), func(t *testing.T) {
57	s, g, _, model := doorPlanFixture(t, true)
58	const offset = 4 << 20
59	payload := append(bytes.Repeat([]byte("p"), offset), []byte("original")...)
60	key := dbVersionKey(g.claim.Claim.Holder.Location, 0, 1, 1)
61	id := model.Put("data", key, payload)
62	model.Put("data", key, []byte("replacement"))
63	var read atomic.Int64
64	base := doorOpenHook{VersionedStore: model, open: func(ctx context.Context, b, k, v string) (io.ReadCloser, error) {
65	rc, err := model.OpenVersion(ctx, b, k, v)
66	return &doorCountedReader{ReadCloser: rc, read: &read}, err
67	}}
68	var store lease.VersionedStore = base
69	if ranged {
70	store = doorRangeHook{doorOpenHook: base, openRange: func(ctx context.Context, b, k, v string, off, length int64) (io.ReadCloser, error) {
71	if b != "data" || k != key || v != id || off != offset || length != 3 {
72	t.Fatalf("range arguments = %s %s %s %d %d", b, k, v, off, length)
73	}
74	rc, err := model.OpenVersionRange(ctx, b, k, v, off, length)
75	return &doorCountedReader{ReadCloser: rc, read: &read}, err
76	}}
77	}
78	rd := &dbVersionReader{store: store, bucket: s.cfg.HibernateBucket, handle: dbFileHandle{
79	Location: g.claim.Claim.Holder.Location, Key: key, VersionID: id, Size: int64(len(payload)),
80	}}
81	rc, err := rd.OpenLTXFile(t.Context(), 0, 1, 1, offset, 3)
82	if err != nil {
83	t.Fatal(err)
84	}
85	body, err := io.ReadAll(rc)
86	rc.Close()
87	if err != nil || string(body) != "ori" {
88	t.Fatalf("body = %q, %v", body, err)
89	}
90	want := int64(offset + 3)
91	if ranged {
92	want = 3
93	}
94	if got := read.Load(); got != want {
95	t.Fatalf("storage read %d bytes; want %d", got, want)
96	}
97	})
98	}
99	}
100
101	func TestDoorPlanRangeErrorsNeverFallBack(t *testing.T) {
102	for _, failure := range []error{lease.ErrMirrorNotFound, lease.ErrIsDeleteMarker, context.Canceled, errors.New("offline")} {
103	t.Run(failure.Error(), func(t *testing.T) {
104	s, g, _, model := doorPlanFixture(t, true)
105	key := dbVersionKey(g.claim.Claim.Holder.Location, 0, 1, 1)
106	model.Put("data", key, []byte("replacement"))
107	var fullOpens int
108	store := doorRangeHook{
109	doorOpenHook: doorOpenHook{VersionedStore: model, open: func(context.Context, string, string, string) (io.ReadCloser, error) {
110	fullOpens++
111	return io.NopCloser(bytes.NewReader([]byte("fallback"))), nil
112	}},
113	openRange: func(context.Context, string, string, string, int64, int64) (io.ReadCloser, error) {
114	return nil, failure
115	},
116	}
117	rd := &dbVersionReader{store: store, bucket: s.cfg.HibernateBucket, handle: dbFileHandle{
118	Location: g.claim.Claim.Holder.Location, Key: key, VersionID: "selected", Size: 8,
119	}}
120	rc, err := rd.OpenLTXFile(t.Context(), 0, 1, 1, 2, 3)
121	if rc != nil {
122	rc.Close()
123	}
124	want := failure
125	if errors.Is(failure, lease.ErrMirrorNotFound) {
126	want = os.ErrNotExist
127	}
128	if !errors.Is(err, want) || fullOpens != 0 {
129	t.Fatalf("range error = %v; want %v, full opens = %d", err, want, fullOpens)
130	}
131	})
132	}
133	}
134
135	func TestDoorPlanHTTPDeepRangeKeepsWatchdogAlive(t *testing.T) {
136	fx := dbFixture(t)
137	model := enableDoorPlan(t, fx)
138	const offset = 4 << 20
139	want := bytes.Repeat([]byte("tail"), 64<<10)
140	payload := append(bytes.Repeat([]byte("p"), offset), want...)
141	copy(payload, doorLTXHeader(t, 1, 1))
142	key := dbVersionKey("instances/"+dbTestHost, 0, 1, 1)
143	id := model.Put("data-test", key, payload)
144	var read, fullOpens, rangeOpens atomic.Int64
145	fx.srv.dbStreamIdle = 250 * time.Millisecond
146	fx.srv.cfg.DBVersions = doorRangeHook{
147	doorOpenHook: doorOpenHook{VersionedStore: model, open: func(ctx context.Context, b, k, v string) (io.ReadCloser, error) {
148	fullOpens.Add(1)
149	rc, err := model.OpenVersion(ctx, b, k, v)
150	return &doorCountedReader{ReadCloser: rc, ctx: ctx, read: &read, delay: 10 * time.Millisecond}, err
151	}},
152	openRange: func(ctx context.Context, b, k, v string, off, length int64) (io.ReadCloser, error) {
153	rangeOpens.Add(1)
154	if v != id || off != offset || length != int64(len(want)) {
155	return nil, fmt.Errorf("unexpected range %s %d %d", v, off, length)
156	}
157	rc, err := model.OpenVersionRange(ctx, b, k, v, off, length)
158	return &doorCountedReader{ReadCloser: rc, ctx: ctx, read: &read, delay: 10 * time.Millisecond}, err
159	},
160	}
161	status, _, body := rawReq(t, http.MethodGet, fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0)), fx.member, "")
162	var list dbdoor.ListResponse
163	if err := json.Unmarshal([]byte(body), &list); err != nil || status != 200 || len(list.Files) != 1 {
164	t.Fatalf("list = %d %s, %v", status, body, err)
165	}
166	read.Store(0)
167	fullOpens.Store(0)
168	model.Put("data-test", key, []byte("replacement"))
169	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
170	defer cancel()
171	req, err := http.NewRequestWithContext(ctx, http.MethodGet, fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 1, 1)), nil)
172	if err != nil {
173	t.Fatal(err)
174	}
175	req.Header.Set("Authorization", "Bearer "+fx.member)
176	req.Header.Set(dbdoor.HandleHeader, list.Files[0].Handle)
177	req.Header.Set("Range", fmt.Sprintf("bytes=%d-", offset))
178	start := time.Now()
179	res, err := http.DefaultClient.Do(req)
180	if err != nil {
181	t.Fatal(err)
182	}
183	got, err := io.ReadAll(res.Body)
184	res.Body.Close()
185	if err != nil || res.StatusCode != 206 || !bytes.Equal(got, want) {
186	t.Fatalf("range = status %d, %d bytes, %v; want 206 and %d bytes", res.StatusCode, len(got), err, len(want))
187	}
188	if time.Since(start) <= 2*fx.srv.dbStreamIdle {
189	t.Fatal("stream did not outlive two watchdog intervals")
190	}
191	if read.Load() != int64(len(want)) || fullOpens.Load() != 0 || rangeOpens.Load() != 1 {
192	t.Fatalf("read %d bytes, full opens %d, range opens %d", read.Load(), fullOpens.Load(), rangeOpens.Load())
193	}
194	if res.Header.Get("Content-Range") != fmt.Sprintf("bytes %d-%d/%d", offset, len(payload)-1, len(payload)) || res.Header.Get(dbdoor.HandleHeader) != list.Files[0].Handle {
195	t.Fatalf("range headers = %v", res.Header)
196	}
197	fx.waitIdle(t, time.Second)
198	}
199