carlos / platform Public

Download this file

1	package rootparticipation
2
3	import (
4	"bytes"
5	"context"
6	"encoding/json"
7	"errors"
8	"io"
9	"os"
10	"strings"
11
12	"github.com/carlosframework/platform/internal/rootstate"
13	)
14
15	const Namespace = "/var/lib/carlos-claims/authority/exec"
16
17	var ErrUnavailable = errors.New("root participation unavailable")
18
19	type Census struct {
20	Schema int
21	Box, Enrollment string
22	Migration, Revision, LaunchRevision uint64
23	Complete, NeverEnrolled bool
24	Pending string
25	}
26
27	type Ledger struct {
28	State rootstate.Store
29	Box, Enrollment string
30	VerifyLaunches func(context.Context, uint64) error
31	ExcludeLaunches func(context.Context, func(context.Context) error) error
32	}
33
34	func Decode(raw []byte, v any) error {
35	if len(raw) > rootstate.MaxRecord {
36	return ErrUnavailable
37	}
38	d := json.NewDecoder(bytes.NewReader(raw))
39	d.DisallowUnknownFields()
40	if err := d.Decode(v); err != nil {
41	return err
42	}
43	if d.Decode(new(any)) != io.EOF {
44	return ErrUnavailable
45	}
46	return nil
47	}
48	func (l Ledger) Census() (Census, error) { return l.readCensus(false) }
49
50	func (l Ledger) RecoveryCensus() (Census, error) { return l.readCensus(true) }
51
52	func (l Ledger) readCensus(recovery bool) (Census, error) {
53	var c Census
54	if l.State == nil {
55	return c, ErrUnavailable
56	}
57	raw, e := l.State.Read("census.json")
58	if e != nil {
59	return c, e
60	}
61	if Decode(raw, &c) != nil || c.Schema != 1 || c.Revision == 0 || c.Box == "" || c.Enrollment == "" || !c.Complete || (c.Pending != "" && (!recovery || c.Pending != "acquisition")) || (c.Migration == 0 && !c.NeverEnrolled) || c.LaunchRevision == 0 || (l.Box != "" && c.Box != l.Box) || (l.Enrollment != "" && c.Enrollment != l.Enrollment) {
62	return c, ErrUnavailable
63	}
64	return c, nil
65	}
66	func (l Ledger) writeCensus(c Census) error {
67	raw, e := json.Marshal(c)
68	if e != nil {
69	return e
70	}
71	return l.State.Write("census.json", raw)
72	}
73	func (l Ledger) Acquire(ctx context.Context, commit func() error) error {
74	if l.State == nil || commit == nil {
75	return ErrUnavailable
76	}
77	unlock, e := l.lockTransaction(ctx)
78	if e != nil {
79	return e
80	}
81	defer unlock()
82	c, e := l.Census()
83	if e != nil {
84	return e
85	}
86	if c.NeverEnrolled {
87	return ErrUnavailable
88	}
89	if _, e = l.State.Read("update.json"); !errors.Is(e, os.ErrNotExist) {
90	return ErrUnavailable
91	}
92	c.Pending = "acquisition"
93	c.Revision++
94	if e = l.writeCensus(c); e != nil {
95	return e
96	}
97	e = commit()
98	c.Pending = ""
99	return errors.Join(e, l.writeCensus(c))
100	}
101	func (l Ledger) idle(ctx context.Context, c Census) error {
102	if l.VerifyLaunches == nil {
103	return ErrUnavailable
104	}
105	names, e := l.State.Names()
106	if e != nil {
107	return e
108	}
109	for _, name := range names {
110	if name == "census.json" || name == "update.json" || strings.HasSuffix(name, ".lock") || strings.HasSuffix(name, ".live") {
111	continue
112	}
113	if !strings.HasSuffix(name, ".json") {
114	return ErrUnavailable
115	}
116	raw, e := l.State.Read(name)
117	if e != nil {
118	return e
119	}
120	var r ExecutionRecord
121	if Decode(raw, &r) != nil || r.Schema != 1 || !r.Request.Valid() || r.Request.Host+".json" != name || r.Authority.Box != c.Box || r.Authority.Enrollment != c.Enrollment || r.Authority.Migration != c.Migration || r.Status != "settled" || r.Token == "" || r.Replay == nil {
122	return ErrUnavailable
123	}
124	}
125	return l.VerifyLaunches(ctx, c.LaunchRevision)
126	}
127	func (l Ledger) Idle(ctx context.Context) error {
128	if l.State == nil {
129	return ErrUnavailable
130	}
131	unlock, e := l.lockTransaction(ctx)
132	if e != nil {
133	return e
134	}
135	defer unlock()
136	c, e := l.Census()
137	if e != nil {
138	return e
139	}
140	return l.idle(ctx, c)
141	}
142	func (l Ledger) BeginUpdate(ctx context.Context, id string) error {
143	return l.BeginUpdateWith(ctx, id, nil)
144	}
145
146	func (l Ledger) BeginUpdateWith(ctx context.Context, id string, commit func() error) error {
147	if l.State == nil || !validID(id) {
148	return ErrUnavailable
149	}
150	unlock, e := l.lockTransaction(ctx)
151	if e != nil {
152	return e
153	}
154	defer unlock()
155	c, e := l.Census()
156	if e != nil {
157	return e
158	}
159	if _, e = l.State.Read("update.json"); !errors.Is(e, os.ErrNotExist) {
160	return ErrUnavailable
161	}
162	raw, _ := json.Marshal(struct {
163	Schema int
164	ID string
165	Revision uint64
166	}{1, id, c.Revision})
167	if e = l.State.Write("update.json", raw); e != nil {
168	return e
169	}
170	if e = l.idle(ctx, c); e != nil {
171	return errors.Join(e, l.State.Remove("update.json"))
172	}
173	if commit != nil {
174	return commit()
175	}
176	return nil
177	}
178
179	func (l Ledger) EndUpdateWith(ctx context.Context, id string, commit func() error) error {
180	if l.State == nil {
181	return ErrUnavailable
182	}
183	unlock, e := l.lockTransaction(ctx)
184	if e != nil {
185	return e
186	}
187	defer unlock()
188	raw, e := l.State.Read("update.json")
189	if e != nil {
190	return e
191	}
192	var u struct {
193	Schema int
194	ID string
195	Revision uint64
196	}
197	if Decode(raw, &u) != nil || u.Schema != 1 || u.ID != id {
198	return ErrUnavailable
199	}
200	c, e := l.Census()
201	if e != nil {
202	return e
203	}
204	if c.Revision != u.Revision {
205	return ErrUnavailable
206	}
207	if e = l.idle(ctx, c); e != nil {
208	return e
209	}
210	if commit != nil {
211	if e = commit(); e != nil {
212	return e
213	}
214	}
215	return l.State.Remove("update.json")
216	}
217
218	// HoldUpdate is the acquisition barrier recovery leaves when it defers a
219	// settlement its census refused: root's update record under id, written
220	// under the transaction lock, which refuses every new acquisition. Unlike
221	// BeginUpdate it runs no idle check, since it only adds a refusal and
222	// starts nothing; EndUpdateWith, with its census, is still the only way
223	// the record goes. An existing record is kept as it is, and a census with
224	// an acquisition pending already refuses acquisitions, so neither writes.
225	func (l Ledger) HoldUpdate(ctx context.Context, id string) error {
226	if l.State == nil || !validID(id) {
227	return ErrUnavailable
228	}
229	unlock, e := l.lockTransaction(ctx)
230	if e != nil {
231	return e
232	}
233	defer unlock()
234	c, e := l.RecoveryCensus()
235	if e != nil {
236	return e
237	}
238	if c.Pending != "" {
239	return nil
240	}
241	if _, e = l.State.Read("update.json"); !errors.Is(e, os.ErrNotExist) {
242	return e
243	}
244	raw, _ := json.Marshal(struct {
245	Schema int
246	ID string
247	Revision uint64
248	}{1, id, c.Revision})
249	return l.State.Write("update.json", raw)
250	}
251
252	func (l Ledger) UpdateID() (string, error) {
253	if l.State == nil {
254	return "", ErrUnavailable
255	}
256	raw, err := l.State.Read("update.json")
257	if err != nil {
258	return "", err
259	}
260	var u struct {
261	Schema int
262	ID string
263	Revision uint64
264	}
265	if Decode(raw, &u) != nil || u.Schema != 1 || !validID(u.ID) || u.Revision == 0 {
266	return "", ErrUnavailable
267	}
268	return u.ID, nil
269	}
270
271	func (l Ledger) RecoverAcquisition(ctx context.Context) error {
272	if l.State == nil {
273	return ErrUnavailable
274	}
275	unlock, err := l.lockTransaction(ctx)
276	if err != nil {
277	return err
278	}
279	defer unlock()
280	c, err := l.RecoveryCensus()
281	if err != nil {
282	return err
283	}
284	if c.Pending == "" {
285	return nil
286	}
287	if err := l.idle(context.WithValue(ctx, acquisitionRecoveryKey{}, true), c); err != nil {
288	return err
289	}
290	c.Pending = ""
291	return l.writeCensus(c)
292	}
293
294	type acquisitionRecoveryKey struct{}
295
296	func IsAcquisitionRecovery(ctx context.Context) bool {
297	return ctx.Value(acquisitionRecoveryKey{}) == true
298	}
299