carlos / platform Public

Download this file

1	package dbreads
2
3	import (
4	"context"
5	"errors"
6	"math/rand/v2"
7	"sync"
8	"testing"
9	"time"
10
11	"github.com/carlosframework/platform/internal/provision"
12	"github.com/carlosframework/platform/internal/release"
13	)
14
15	// The identity check (spec §8.2): the record as the envelope named it, or a
16	// code. T9's stages are Task 11's; this is the check itself.
17	func TestCheckIsTheRecordAsTheEnvelopeNamedIt(t *testing.T) {
18	t.Parallel()
19	no := false
20	for name, c := range map[string]struct {
21	change func(fx *fixture)
22	want string
23	}{
24	"unchanged": {func(*fixture) {}, ""},
25	"moved to another region": {func(fx *fixture) { r := fx.record(); r.Region = "ie"; fx.putRecord(r) }, CodeRecordChanged},
26	"moving": {func(fx *fixture) {
27	r := fx.record()
28	r.Move = &provision.MoveState{ID: "mv_x", To: "ie", State: "requested"}
29	fx.putRecord(r)
30	}, CodeMoving},
31	"recreated in the account": {func(fx *fixture) { r := fx.record(); r.CreatedAt = r.CreatedAt.Add(time.Hour); fx.putRecord(r) }, CodeRecordChanged},
32	"recreated in another account": {func(fx *fixture) {
33	created := fx.record().CreatedAt
34	fx.store.Delete(context.Background(), provision.RecordKey(fxAccount, fxApp, fxHost))
35	fx.putRecordUnder("bac", provision.Record{Region: "uk", CreatedAt: created})
36	}, CodeRecordChanged},
37	"no longer sleeps": {func(fx *fixture) { r := fx.record(); r.Hibernate = &no; fx.putRecord(r) }, CodeRecordChanged},
38	"pooled": {func(fx *fixture) { r := fx.record(); r.Pool = "shared"; fx.putRecord(r) }, CodeRecordChanged},
39	"placed on a fleet": {func(fx *fixture) { r := fx.record(); r.Region = "fleet/acme"; fx.putRecord(r) }, CodeRecordChanged},
40	} {
41	t.Run(name, func(t *testing.T) {
42	t.Parallel()
43	fx := newFixture(t, 0)
44	req := fx.request("SELECT 1", "")
45	c.change(fx)
46	a, _ := fx.agent("uk-1")
47	_, code, err := a.check(context.Background(), req)
48	if err != nil || code != c.want {
49	t.Fatalf("check = %q, %v; want %q", code, err, c.want)
50	}
51	})
52	}
53	}
54
55	// T5: two capable boxes in one region; exactly one owns each read, across
56	// 1,000 reads with random jitter.
57	func TestElectionOneOwnerAcrossTwoAgents(t *testing.T) {
58	t.Parallel()
59	fx := newFixture(t, 0)
60	a, _ := fx.agent("uk-1")
61	b, _ := fx.agent("uk-2")
62	var mu sync.Mutex
63	owners := map[string]int{}
64	var wg sync.WaitGroup
65	for i := 0; i < 1000; i++ {
66	id, err := MintID()
67	if err != nil {
68	t.Fatal(err)
69	}
70	for _, ag := range []*Agent{a, b} {
71	wg.Add(1)
72	go func(ag *Agent) {
73	defer wg.Done()
74	time.Sleep(rand.N(2 * time.Millisecond))
75	if _, ok := ag.elect(context.Background(), id); ok {
76	mu.Lock()
77	owners[id]++
78	mu.Unlock()
79	}
80	}(ag)
81	}
82	}
83	wg.Wait()
84	if len(owners) != 1000 {
85	t.Fatalf("%d reads have an owner, want 1000", len(owners))
86	}
87	for id, n := range owners {
88	if n != 1 {
89	t.Fatalf("%s has %d owners", id, n)
90	}
91	}
92	}
93
94	// lossyStore commits some Puts and then answers an error: a response lost
95	// on the way back (T5, codex 16).
96	type lossyStore struct {
97	release.Store
98	lose func(key string) bool
99	refuse func(key string) bool
100	}
101
102	func (s lossyStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
103	if s.refuse != nil && s.refuse(key) {
104	return "", errors.New("connection reset before the write")
105	}
106	etag, err := s.Store.Put(ctx, key, body, ifMatch)
107	if err == nil && s.lose != nil && s.lose(key) {
108	return "", errors.New("connection reset after the write")
109	}
110	return etag, err
111	}
112
113	// A box whose picked create commits but answers an error reads it back and
114	// owns the read.
115	func TestElectionReadsBackALostAnswer(t *testing.T) {
116	t.Parallel()
117	fx := newFixture(t, 0)
118	a, _ := fx.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) {
119	c.Store = lossyStore{Store: fx.store, lose: func(string) bool { return true }}
120	})
121	if _, ok := a.elect(context.Background(), "dbr_aaaaaaaaaaaaaaaaaaaaaaaaaa"); !ok {
122	t.Fatal("a lost answer to a create that landed was not owned")
123	}
124	}
125
126	// A box whose create answers an error while another box owns the read steps
127	// away; so does one whose read-back fails.
128	func TestElectionStepsAwayFromAnotherOwner(t *testing.T) {
129	t.Parallel()
130	fx := newFixture(t, 0)
131	a, _ := fx.agent("uk-1")
132	id := "dbr_aaaaaaaaaaaaaaaaaaaaaaaaaa"
133	if _, ok := a.elect(context.Background(), id); !ok {
134	t.Fatal("the first create did not own")
135	}
136	b, _ := fx.agent("uk-2", func(c *AgentConfig, _ *inProcWorkers) {
137	c.Store = lossyStore{Store: fx.store, refuse: func(string) bool { return true }}
138	})
139	if _, ok := b.elect(context.Background(), id); ok {
140	t.Fatal("a box owned a read another box had picked")
141	}
142	c, _ := fx.agent("uk-3", func(cfg *AgentConfig, _ *inProcWorkers) {
143	cfg.Store = failingGets{lossyStore{Store: fx.store, refuse: func(string) bool { return true }}}
144	})
145	if _, ok := c.elect(context.Background(), "dbr_bbbbbbbbbbbbbbbbbbbbbbbbbb"); ok {
146	t.Fatal("a box owned a read it could not read back")
147	}
148	}
149
150	type failingGets struct{ release.Store }
151
152	func (failingGets) Get(context.Context, string) ([]byte, string, error) {
153	return nil, "", errors.New("timeout")
154	}
155