carlos / platform Public

Download this file

1	//go:build linux
2
3	package hostexec
4
5	import (
6	"context"
7	"encoding/json"
8	"fmt"
9	"os"
10	"path/filepath"
11	"testing"
12
13	"golang.org/x/sys/unix"
14	)
15
16	func (d *brokerDisk) HoldLive(name string) (func(), error) {
17	f, err := os.OpenFile(filepath.Join(d.dir, name), os.O_CREATE|os.O_RDWR, 0600)
18	if err != nil {
19	return nil, err
20	}
21	lock := unix.Flock_t{Type: unix.F_WRLCK}
22	if err := unix.FcntlFlock(f.Fd(), unix.F_OFD_SETLK, &lock); err != nil {
23	f.Close()
24	return nil, err
25	}
26	return func() { f.Close() }, nil
27	}
28	func (d *brokerDisk) LiveHeld(name string) (bool, error) {
29	f, err := os.OpenFile(filepath.Join(d.dir, name), os.O_RDWR, 0)
30	if err != nil {
31	return false, err
32	}
33	defer f.Close()
34	lock := unix.Flock_t{Type: unix.F_RDLCK}
35	err = unix.FcntlFlock(f.Fd(), unix.F_OFD_GETLK, &lock)
36	return lock.Type == unix.F_WRLCK && lock.Pid == -1, err
37	}
38
39	type pausedProbe struct {
40	*brokerDisk
41	entered, resume chan struct{}
42	mutate bool
43	}
44
45	func (d *pausedProbe) LiveHeld(name string) (bool, error) {
46	if d.mutate {
47	unlock, err := d.Lock(context.Background(), name)
48	if err != nil {
49	return false, err
50	}
51	close(d.entered)
52	<-d.resume
53	unlock()
54	return false, nil
55	}
56	held, err := d.brokerDisk.LiveHeld(name)
57	close(d.entered)
58	<-d.resume
59	return held, err
60	}
61
62	type falseBrokerProbe struct{ *brokerDisk }
63
64	func (d falseBrokerProbe) LiveHeld(name string) (bool, error) {
65	f, err := os.OpenFile(filepath.Join(d.dir, name), os.O_RDWR, 0)
66	if err != nil {
67	return false, err
68	}
69	defer f.Close()
70	err = unix.Flock(int(f.Fd()), unix.LOCK_EX|unix.LOCK_NB)
71	if err == unix.EWOULDBLOCK {
72	return true, nil
73	}
74	return false, err
75	}
76
77	func TestPlan14ConcurrentReadersNeverGrantLiveness(t *testing.T) {
78	for _, mutate := range []bool{false, true} {
79	t.Run(fmt.Sprint(mutate), func(t *testing.T) {
80	b, d, _, clock, q, oldPeer := brokerFixture(t)
81	peer, err := currentControllerProcess()
82	if err != nil {
83	t.Fatal(err)
84	}
85	b.cfg.BootID = peer.KernelBootID
86	original := b.cfg.Authority
87	b.cfg.Authority = func(ctx context.Context, q ExecutionRequest, _ ControllerProcess) (ExecutionAuthority, error) {
88	return original(ctx, q, oldPeer)
89	}
90	if _, err = b.Execute(t.Context(), peer, q); err != nil {
91	t.Fatal(err)
92	}
93	census, _ := json.Marshal(map[string]any{"Schema": 1, "Box": "box", "Enrollment": "en", "Migration": 1, "Revision": 1, "LaunchRevision": 1, "Complete": true})
94	if err = d.Write("census.json", census); err != nil {
95	t.Fatal(err)
96	}
97	if _, err = ReadRoot(t.Context(), d, clock, q, false); err != nil {
98	t.Fatal(err)
99	}
100	raw, err := d.Read(q.Host + ".json")
101	if err != nil {
102	t.Fatal(err)
103	}
104	r, err := b.read(q.Host)
105	if err != nil {
106	t.Fatal(err)
107	}
108	d.failAt = d.writes + 1
109	if err = b.deny(t.Context(), r, ErrLost); err == nil {
110	t.Fatal("denial succeeded")
111	}
112	if err = os.WriteFile(filepath.Join(d.dir, q.Host+".json"), raw, 0600); err != nil {
113	t.Fatal(err)
114	}
115	a := &pausedProbe{brokerDisk: d, entered: make(chan struct{}), resume: make(chan struct{}), mutate: mutate}
116	done := make(chan error, 1)
117	go func() { _, err := ReadRoot(t.Context(), a, clock, q, false); done <- err }()
118	<-a.entered
119	var readErr error
120	if mutate {
121	_, readErr = ReadRoot(t.Context(), falseBrokerProbe{d}, clock, q, false)
122	} else {
123	_, readErr = ReadRoot(t.Context(), d, clock, q, false)
124	}
125	close(a.resume)
126	if err := <-done; err == nil {
127	t.Fatal("reader A admitted")
128	}
129	if (readErr == nil) != mutate {
130	t.Fatalf("mutation=%v B error=%v", mutate, readErr)
131	}
132	})
133	}
134	}
135