carlos / platform Public

Download this file

1	package registry
2
3	import (
4	"bytes"
5	"context"
6	"database/sql"
7	"encoding/json"
8	"errors"
9	)
10
11	type ColdRows struct {
12	Route Route
13	Timers []Timer
14	}
15
16	// canonicalJSON decodes stored bytes into a T and encodes them again with
17	// this binary. Bytes written by an older binary lack any field added since,
18	// so they decode with those fields empty and re-encode in today's shape: an
19	// unchanged row then compares equal to a fresh encoding, while a row whose
20	// newer fields have since been set still differs. This is the only
21	// normalisation applied before the ownership comparisons below. An
22	// undecodable body returns an error, and the caller keeps its fail-closed
23	// ErrClaimOwnership.
24	func canonicalJSON[T any](stored []byte) ([]byte, error) {
25	var v T
26	if err := json.Unmarshal(stored, &v); err != nil {
27	return nil, err
28	}
29	return json.Marshal(v)
30	}
31
32	// withoutBuildStateCopy is rt with its build_state column cleared, for the
33	// claim handovers' row-identity checks (RestoreClaimRows' replay,
34	// RebindClaimRows). The column is the agent's copy of the bucket's
35	// build-state object, rewritten at every reconcile refresh even while the
36	// instance is stopped (its read_at moves; snapshotChurnCols). A refresh
37	// between a handover's read and its transaction, or before its replay, is
38	// not a change to the installation, and refusing on it would leave a
39	// handover that lost its reply unable ever to replay. Neither check writes
40	// the column from the caller's copy on those paths.
41	func withoutBuildStateCopy(rt Route) Route {
42	rt.BuildState = ""
43	return rt
44	}
45
46	// coldRowsWithoutBuildStateCopy is c with its route's build_state cleared
47	// (withoutBuildStateCopy).
48	func coldRowsWithoutBuildStateCopy(c ColdRows) ColdRows {
49	c.Route = withoutBuildStateCopy(c.Route)
50	return c
51	}
52
53	func (r *Registry) RestoreClaimRows(ctx context.Context, id, retirement string, old ColdRows, next ColdRows) error {
54	if r.remote != nil {
55	_, err := r.remote.call(ctx, remoteRequest{Method: "RestoreClaimRows", Id: id, Retirement: retirement, Old: old, NextValue: next}, nil)
56	return err
57	}
58
59	if id == "" || retirement == "" || !old.Route.ClaimOwnership.Valid() || !next.Route.ClaimOwnership.Valid() || old.Route.Host != next.Route.Host || old.Route.ClaimIncarnation != next.Route.ClaimIncarnation || next.Route.ClaimStamp <= old.Route.ClaimStamp || len(old.Timers) != len(next.Timers) {
60	return ErrClaimOwnership
61	}
62	before, after := old.Route, next.Route
63	before.ClaimOwnership = ClaimOwnership{}
64	after.ClaimOwnership = ClaimOwnership{}
65	if before != after {
66	return ErrClaimOwnership
67	}
68	for i, t := range next.Timers {
69	a, b := old.Timers[i], t
70	a.ClaimOwnership = ClaimOwnership{}
71	b.ClaimOwnership = ClaimOwnership{}
72	if a != b || !t.ClaimOwnership.Valid() || t.Host != next.Route.Host || t.ClaimIncarnation != next.Route.ClaimIncarnation || t.ClaimStamp != next.Route.ClaimStamp || t.ClaimOwner != next.Route.ClaimOwner {
73	return ErrClaimOwnership
74	}
75	}
76	if _, err := r.sql.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS cold_handovers(id TEXT PRIMARY KEY,body BLOB NOT NULL)`); err != nil {
77	return err
78	}
79	tx, err := r.beginClaimTransaction(ctx, true)
80	if err != nil {
81	return err
82	}
83	defer tx.Rollback()
84	encoded, err := json.Marshal(next)
85	if err != nil {
86	return err
87	}
88	var saved []byte
89	err = tx.QueryRowContext(ctx, `SELECT body FROM cold_handovers WHERE id=?`, id).Scan(&saved)
90	if err == nil {
91	// A body saved by a binary without today's Route fields is
92	// re-encoded first, or a handover started before an upgrade could
93	// never be replayed after it.
94	canonical, e := canonicalJSON[ColdRows](saved)
95	if e != nil || !bytes.Equal(canonical, encoded) {
96	return ErrClaimOwnership
97	}
98	current, e := scanRoute(tx.QueryRowContext(ctx, `SELECT `+routeCols+` FROM routes WHERE host=?`, next.Route.Host).Scan)
99	if e != nil || withoutBuildStateCopy(current) != withoutBuildStateCopy(next.Route) {
100	return ErrClaimOwnership
101	}
102	for _, want := range next.Timers {
103	got, e := scanTimer(tx.QueryRowContext(ctx, `SELECT `+timerCols+` FROM timers WHERE host=? AND name=?`, want.Host, want.Name))
104	if e != nil || got != want {
105	return ErrClaimOwnership
106	}
107	}
108	return nil
109	}
110	if !errors.Is(err, sql.ErrNoRows) {
111	return err
112	}
113	verify := func(kind, host, name, rowID string, want any) error {
114	var raw []byte
115	if err := tx.QueryRowContext(ctx, `SELECT archive FROM claim_row_retirements WHERE journal=? AND kind=? AND host=? AND name=? AND row_id=?`, retirement, kind, host, name, rowID).Scan(&raw); err != nil {
116	return err
117	}
118	var archive ClaimRowArchive
119	if json.Unmarshal(raw, &archive) != nil {
120	return ErrClaimOwnership
121	}
122	contents, err := json.Marshal(want)
123	if err != nil {
124	return err
125	}
126	stored := []byte(archive.Contents)
127	if kind == "route" {
128	// An archive retired by a binary without today's Route
129	// fields is re-encoded before the comparison (see canonicalJSON).
130	if stored, err = canonicalJSON[Route](stored); err != nil {
131	return ErrClaimOwnership
132	}
133	}
134	if !bytes.Equal(contents, stored) {
135	return ErrClaimOwnership
136	}
137	return nil
138	}
139	if err = verify("route", old.Route.Host, "", old.Route.ClaimRowID, old.Route); err != nil {
140	return err
141	}
142	for _, t := range old.Timers {
143	if err = verify("timer", t.Host, t.Name, t.ClaimRowID, struct {
144	Timer Timer
145	Registration string
146	}{t, t.Registration}); err != nil {
147	return err
148	}
149	}
150	var count int
151	if err = tx.QueryRowContext(ctx, `SELECT (SELECT COUNT(*) FROM routes WHERE host=?)+(SELECT COUNT(*) FROM timers WHERE host=?)`, next.Route.Host, next.Route.Host).Scan(&count); err != nil {
152	return err
153	}
154	if count != 0 {
155	return ErrClaimOwnership
156	}
157	if _, err = tx.ExecContext(ctx, `INSERT INTO routes (`+routeCols+`) VALUES (`+placeholders(routeColCount)+`)`, routeArgs(next.Route)...); err != nil {
158	return err
159	}
160	for _, t := range next.Timers {
161	if _, err = tx.ExecContext(ctx, `INSERT INTO timers (`+timerCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, t.Host, t.Name, t.Kind, t.Spec, t.Path, t.NextAt, t.LastAt, t.LastStatus, t.LastMs, t.Attempts, t.CreatedAt, t.DueAt, t.Registration, t.ClaimIncarnation, t.ClaimStamp, t.ClaimOwner, t.ClaimRowID); err != nil {
162	return err
163	}
164	}
165	if _, err = tx.ExecContext(ctx, `INSERT INTO cold_handovers(id,body) VALUES (?,?)`, id, encoded); err != nil {
166	return err
167	}
168	return tx.Commit()
169	}
170
171	type coldRowsWire struct {
172	Route Route
173	Timers []coldTimerWire
174	}
175	type coldTimerWire struct {
176	Timer Timer
177	Registration string
178	}
179
180	func (r ColdRows) MarshalJSON() ([]byte, error) {
181	wire := coldRowsWire{Route: r.Route}
182	for _, timer := range r.Timers {
183	wire.Timers = append(wire.Timers, coldTimerWire{timer, timer.Registration})
184	}
185	return json.Marshal(wire)
186	}
187	func (r *ColdRows) UnmarshalJSON(body []byte) error {
188	var wire coldRowsWire
189	if err := json.Unmarshal(body, &wire); err != nil {
190	return err
191	}
192	r.Route = wire.Route
193	r.Timers = nil
194	for _, timer := range wire.Timers {
195	timer.Timer.Registration = timer.Registration
196	r.Timers = append(r.Timers, timer.Timer)
197	}
198	return nil
199	}
200