carlos / platform Public

Download this file

1	package runs
2
3	import (
4	"context"
5	"errors"
6	"io"
7	"strings"
8	"sync"
9	"testing"
10	"time"
11
12	"github.com/carlosframework/platform/internal/release"
13	)
14
15	const tid = "run_abcdefghijklmnopqrstuvwxyz"
16
17	func chunksOf(t *testing.T, s release.Store) []string {
18	t.Helper()
19	keys, err := ListChunks(context.Background(), s, "bab", "shop", tid, 0, 0)
20	if err != nil {
21	t.Fatal(err)
22	}
23	var out []string
24	for _, k := range keys {
25	raw, _, err := s.Get(context.Background(), k)
26	if err != nil {
27	t.Fatal(err)
28	}
29	_, stream, _ := ParseChunkKey(k, OutPrefix("bab", "shop", tid))
30	out = append(out, stream+":"+string(raw))
31	}
32	return out
33	}
34
35	func TestChunkerKeepsReadOrderAcrossStreams(t *testing.T) {
36	s := release.NewDirStore(t.TempDir())
37	c := newChunker(context.Background(), s, "bab", "shop", tid, time.Hour, MaxOutputBytes)
38	io.WriteString(c.Stream(StreamOut), "a")
39	io.WriteString(c.Stream(StreamErr), "b")
40	io.WriteString(c.Stream(StreamOut), "c")
41	io.WriteString(c.Stream(StreamOut), "d")
42	last, truncated, err := c.Close()
43	if err != nil || truncated || last != 3 {
44	t.Fatalf("Close = %d %v %v", last, truncated, err)
45	}
46	if got := strings.Join(chunksOf(t, s), " "); got != "o:a e:b o:cd" {
47	t.Fatalf("chunks %q", got)
48	}
49	}
50
51	func TestChunkerFlushesOnTheTickAndAtTheSizeLimit(t *testing.T) {
52	s := release.NewDirStore(t.TempDir())
53	c := newChunker(context.Background(), s, "bab", "shop", tid, 20*time.Millisecond, MaxOutputBytes)
54	io.WriteString(c.Stream(StreamOut), "tick")
55	deadline := time.Now().Add(2 * time.Second)
56	for len(chunksOf(t, s)) == 0 && time.Now().Before(deadline) {
57	time.Sleep(10 * time.Millisecond)
58	}
59	if got := chunksOf(t, s); len(got) != 1 || got[0] != "o:tick" {
60	t.Fatalf("after a tick: %q", got)
61	}
62	c2 := newChunker(context.Background(), s, "bab", "shop", "run_bbbbbbbbbbbbbbbbbbbbbbbbbb", time.Hour, MaxOutputBytes)
63	io.WriteString(c2.Stream(StreamOut), strings.Repeat("x", ChunkBytes+1))
64	keys, _ := ListChunks(context.Background(), s, "bab", "shop", "run_bbbbbbbbbbbbbbbbbbbbbbbbbb", 0, 0)
65	if len(keys) != 1 {
66	t.Fatalf("a full chunk must flush at once: %q", keys)
67	}
68	if last, _, _ := c2.Close(); last != 2 {
69	t.Fatalf("last %d, want 2 (one full chunk, one byte)", last)
70	}
71	c.Close()
72	}
73
74	func TestChunkerStopsAtTheCapAndSaysSo(t *testing.T) {
75	s := release.NewDirStore(t.TempDir())
76	c := newChunker(context.Background(), s, "bab", "shop", tid, time.Hour, 10)
77	n, err := io.WriteString(c.Stream(StreamOut), "0123456789abcdef")
78	if n != 16 || err != nil {
79	t.Fatalf("a write past the cap must not fail the stream: %d %v", n, err)
80	}
81	io.WriteString(c.Stream(StreamErr), "more")
82	_, truncated, err := c.Close()
83	if !truncated || err != nil {
84	t.Fatalf("truncated %v err %v", truncated, err)
85	}
86	if got := strings.Join(chunksOf(t, s), ""); got != "o:0123456789" {
87	t.Fatalf("chunks %q", got)
88	}
89	}
90
91	// failingStore fails the first n Puts.
92	type failingStore struct {
93	release.Store
94	mu sync.Mutex
95	fail int
96	}
97
98	func (f *failingStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
99	f.mu.Lock()
100	if f.fail > 0 {
101	f.fail--
102	f.mu.Unlock()
103	return "", errors.New("slow down")
104	}
105	f.mu.Unlock()
106	return f.Store.Put(ctx, key, body, ifMatch)
107	}
108
109	func TestChunkerRetriesAFailedPut(t *testing.T) {
110	s := &failingStore{Store: release.NewDirStore(t.TempDir()), fail: 2}
111	c := newChunker(context.Background(), s, "bab", "shop", tid, time.Hour, MaxOutputBytes)
112	io.WriteString(c.Stream(StreamOut), "once")
113	last, _, err := c.Close()
114	if err != nil || last != 1 {
115	t.Fatalf("Close = %d %v", last, err)
116	}
117	if got := chunksOf(t, s.Store); len(got) != 1 || got[0] != "o:once" {
118	t.Fatalf("chunks %q", got)
119	}
120	}
121
122	func TestChunkerGivesUpWhenItsContextEnds(t *testing.T) {
123	ctx, cancel := context.WithCancel(context.Background())
124	s := &failingStore{Store: release.NewDirStore(t.TempDir()), fail: 1 << 30}
125	c := newChunker(ctx, s, "bab", "shop", tid, time.Hour, MaxOutputBytes)
126	go func() { time.Sleep(50 * time.Millisecond); cancel() }()
127	io.WriteString(c.Stream(StreamOut), strings.Repeat("x", ChunkBytes))
128	_, truncated, err := c.Close()
129	if err == nil || !truncated {
130	t.Fatalf("a chunk that never lands: truncated %v err %v", truncated, err)
131	}
132	}
133
134	// dyingStore lets the first ok Puts through and fails every one after.
135	type dyingStore struct {
136	release.Store
137	mu sync.Mutex
138	ok int
139	}
140
141	func (d *dyingStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
142	d.mu.Lock()
143	allowed := d.ok > 0
144	if allowed {
145	d.ok--
146	}
147	d.mu.Unlock()
148	if !allowed {
149	return "", errors.New("gone")
150	}
151	return d.Store.Put(ctx, key, body, ifMatch)
152	}
153
154	func TestChunkerReportsTheLastSeqThatLanded(t *testing.T) {
155	ctx, cancel := context.WithCancel(context.Background())
156	s := &dyingStore{Store: release.NewDirStore(t.TempDir()), ok: 1}
157	c := newChunker(ctx, s, "bab", "shop", tid, time.Hour, MaxOutputBytes)
158	go func() { time.Sleep(50 * time.Millisecond); cancel() }()
159	io.WriteString(c.Stream(StreamOut), strings.Repeat("x", ChunkBytes)) // chunk 1 lands
160	io.WriteString(c.Stream(StreamOut), strings.Repeat("y", ChunkBytes)) // chunk 2 gives up
161	last, truncated, err := c.Close()
162	if last != 1 || !truncated || err == nil {
163	t.Fatalf("Close = %d %v %v, want 1 true and an error", last, truncated, err)
164	}
165	if got := chunksOf(t, s.Store); len(got) != 1 {
166	t.Fatalf("chunks %d, want 1", len(got))
167	}
168	}
169
170	// A give-up in the middle of one large write must end the write, not spin
171	// on it holding the lock.
172	func TestChunkerGivingUpMidWriteDoesNotHang(t *testing.T) {
173	ctx, cancel := context.WithCancel(context.Background())
174	cancel()
175	s := &failingStore{Store: release.NewDirStore(t.TempDir()), fail: 1 << 30}
176	c := newChunker(ctx, s, "bab", "shop", tid, time.Hour, MaxOutputBytes)
177	done := make(chan struct{})
178	var last int64
179	var truncated bool
180	var err error
181	go func() {
182	defer close(done)
183	n, werr := c.Stream(StreamOut).Write(make([]byte, 2*ChunkBytes+1))
184	if n != 2*ChunkBytes+1 || werr != nil {
185	t.Errorf("Write = %d %v", n, werr)
186	}
187	last, truncated, err = c.Close()
188	}()
189	select {
190	case <-done:
191	case <-time.After(10 * time.Second):
192	t.Fatal("Write or Close hung after the chunker gave up")
193	}
194	if last != 0 || !truncated || err == nil {
195	t.Fatalf("Close = %d %v %v", last, truncated, err)
196	}
197	}
198
199	func TestChunkerRejoinsEachStreamWhenBothWriteAtOnce(t *testing.T) {
200	s := release.NewDirStore(t.TempDir())
201	c := newChunker(context.Background(), s, "bab", "shop", tid, 2*time.Millisecond, MaxOutputBytes)
202	want := map[string]string{}
203	var wg sync.WaitGroup
204	for _, stream := range []string{StreamOut, StreamErr} {
205	var b strings.Builder
206	for i := 0; i < 2000; i++ {
207	b.WriteString(stream)
208	b.WriteString(strings.Repeat(string(rune('a'+i%26)), i%37))
209	b.WriteString("\n")
210	}
211	data := b.String()
212	want[stream] = data
213	wg.Add(1)
214	go func() {
215	defer wg.Done()
216	w := c.Stream(stream)
217	for len(data) > 0 {
218	n := min(len(data), 1+len(data)%50)
219	w.Write([]byte(data[:n]))
220	data = data[n:]
221	time.Sleep(time.Microsecond)
222	}
223	}()
224	}
225	wg.Wait()
226	if _, truncated, err := c.Close(); truncated || err != nil {
227	t.Fatalf("Close: %v %v", truncated, err)
228	}
229	got := map[string]string{}
230	for _, ch := range chunksOf(t, s) {
231	got[ch[:1]] += ch[2:]
232	}
233	for _, stream := range []string{StreamOut, StreamErr} {
234	if got[stream] != want[stream] {
235	t.Fatalf("stream %s: got %d bytes, want %d, or the bytes differ", stream, len(got[stream]), len(want[stream]))
236	}
237	}
238	}
239
240	func TestChunkerRefusesAnInvalidRunAtConstruction(t *testing.T) {
241	s := release.NewDirStore(t.TempDir())
242	for _, tc := range []struct {
243	acct, app, id string
244	want error
245	}{
246	{"bab", "shop", "run_../../x", ErrBadID},
247	{"bab", "shop", "", ErrBadID},
248	{"bab", "../other", tid, ErrBadApp},
249	{"../bab", "shop", tid, ErrBadApp},
250	} {
251	c := NewChunker(context.Background(), s, tc.acct, tc.app, tc.id)
252	n, err := io.WriteString(c.Stream(StreamOut), "x")
253	if n != 1 || err != nil {
254	t.Fatalf("%+v: a refused chunker still swallows writes: %d %v", tc, n, err)
255	}
256	if _, _, err := c.Close(); !errors.Is(err, tc.want) {
257	t.Fatalf("%+v: Close err %v, want %v", tc, err, tc.want)
258	}
259	}
260	if keys, _ := s.List(context.Background(), ""); len(keys) != 0 {
261	t.Fatalf("a refused chunker wrote %q", keys)
262	}
263	}
264
265	func TestChunkerRefusesAnUnknownStream(t *testing.T) {
266	s := release.NewDirStore(t.TempDir())
267	c := newChunker(context.Background(), s, "bab", "shop", tid, time.Hour, MaxOutputBytes)
268	io.WriteString(c.Stream("../x"), "bad")
269	_, _, err := c.Close()
270	if err == nil {
271	t.Fatal("an unknown stream must surface in Close")
272	}
273	if keys, _ := s.List(context.Background(), ""); len(keys) != 0 {
274	t.Fatalf("wrote %q", keys)
275	}
276	}
277
278	func TestChunkerDropsWritesAfterClose(t *testing.T) {
279	s := release.NewDirStore(t.TempDir())
280	c := newChunker(context.Background(), s, "bab", "shop", tid, time.Hour, MaxOutputBytes)
281	io.WriteString(c.Stream(StreamOut), "a")
282	last, _, _ := c.Close()
283	io.WriteString(c.Stream(StreamOut), "late")
284	last2, _, _ := c.Close()
285	if last != 1 || last2 != 1 {
286	t.Fatalf("last %d then %d", last, last2)
287	}
288	if got := chunksOf(t, s); len(got) != 1 || got[0] != "o:a" {
289	t.Fatalf("chunks %q", got)
290	}
291	}
292