amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package hostbox
2
3import (
4 "context"
5 "database/sql"
6 "errors"
7 "os"
8 "path/filepath"
9 "reflect"
10
11 "github.com/carlosframework/platform/internal/hostclaim"
12 "github.com/carlosframework/platform/internal/hostguard"
13 "github.com/carlosframework/platform/internal/provision"
14 "github.com/carlosframework/platform/internal/registry"
15)
16
17// ServingHandover retains the stopped installation's state directory. Its
18// barrier covers inventory and row transfer; database retirement follows later.
19type ServingHandover struct {
20 ColdHandover
21 Validate func(context.Context, *hostclaim.Claim, hostguard.FenceTarget, ColdSuccessor, StopInventory, hostguard.RootInstallation) error
22 VerifyDirectory func(context.Context, hostguard.FenceTarget, StopInventoryEntry, Artifact) error
23 checkpoint func(string) error
24}
25
26func (h ServingHandover) Run(ctx context.Context, prior hostguard.FenceTarget, next ColdSuccessor) error {
27 if h.Guard.Root == nil || h.Validate == nil || h.Claims == nil || h.Registry == nil || h.Inventory == nil || h.Inventory.Remote() || h.Launches == nil || h.VerifyReceipt == nil || !filepath.IsAbs(h.JournalDir) || h.Roots["data"] != h.Guard.DataDir || prior.Op != "" || prior.Phase != "" || prior.ObligationID != "" || next.Op != "" || next.Phase != "" || next.Host != prior.Host || next.Incarnation != prior.Incarnation || next.Box != prior.Box || next.Enrollment != prior.Enrollment || next.Stamp <= prior.Stamp || next.Session != next.Stamp || next.Session <= prior.Session || next.Generation == "" || next.ETag == "" {
28 return ErrUnsupported
29 }
30 id := "serving-handover-" + ownershipFor(next.Host, next.Incarnation, next.Stamp, next.Box)
31 path := filepath.Join(h.JournalDir, id+".json")
32 var j coldJournal
33 body, err := readTrustedJournal(path)
34 if err == nil {
35 if decodeStrict(body, &j) != nil || j.V != 1 || j.ID != id || j.Prior != prior || j.Successor != next || j.ServingInstallation == nil || (j.Phase != "prepared" && j.Phase != "complete") {
36 return ErrTypedJournal
37 }
38 } else if !errors.Is(err, os.ErrNotExist) {
39 return err
40 } else {
41 captured := hostguard.Captured{Incarnation: prior.Incarnation, Stamp: prior.Stamp}
42 view, err := h.Guard.Root.View(ctx, prior.Host, captured)
43 if err != nil {
44 return err
45 }
46 in, rev, err := h.Inventory.Snapshot(ctx, prior.Host, captured)
47 if err != nil {
48 return err
49 }
50 if view.Installation.Stamp != prior.Stamp || view.Installation.InventoryRevision != rev || !in.Complete || in.Box != prior.Box || in.Enrollment != prior.Enrollment || in.DirectoryRevision != prior.DirectoryRevision {
51 return ErrTypedInventory
52 }
53 var receipt hostguard.FenceReceipt
54 if prior.Session != 0 {
55 receipt, err = h.Guard.FenceReceiptFor(ctx, prior)
56 if err != nil {
57 return err
58 }
59 if err = h.VerifyReceipt(ctx, receipt); err != nil {
60 return err
61 }
62 }
63 row, err := h.Registry.Get(prior.Host)
64 if err != nil {
65 return err
66 }
67 if row == nil || row.Kind != "instance" || row.Account != view.Installation.Account || row.App != view.Installation.App || row.Socket != provision.NestedSocket(prior.Host) || row.DBPath != provision.NestedDBPath(h.Roots["data"], prior.Host) || row.Unit != "" || row.Addr != "" || row.Addrs != "" || row.Backing != "" || row.Tunnel != "" || row.Platform || row.Canary || row.Replica {
68 return ErrTypedInventory
69 }
70 timers, err := h.Registry.TimersFor(prior.Host)
71 if err != nil {
72 return err
73 }
74 before := registry.ColdRows{Route: *row, Timers: timers}
75 after := registry.ColdRows{Route: *row, Timers: append([]registry.Timer(nil), timers...)}
76 owner := ownershipFor(next.Host, next.Incarnation, next.Stamp, next.Box)
77 after.Route.ClaimOwnership = registry.ClaimOwnership{ClaimIncarnation: next.Incarnation, ClaimStamp: next.Stamp, ClaimOwner: owner, ClaimRowID: rowIDFor(owner, "route", next.Host)}
78 for i := range after.Timers {
79 after.Timers[i].ClaimOwnership = after.Route.ClaimOwnership
80 after.Timers[i].ClaimRowID = rowIDFor(owner, "timer", after.Timers[i].Name)
81 }
82 barrier := prior
83 barrier.FenceID, barrier.Stamp, barrier.Session, barrier.ClaimETag, barrier.DirectoryRevision = id, next.Stamp, next.Session, next.ETag, next.DirectoryRevision
84 j = coldJournal{V: 1, ID: id, Phase: "prepared", Prior: prior, Barrier: barrier, Successor: next, Inventory: in, Receipt: receipt, Before: before, After: after, ServingInstallation: &view.Installation}
85 err = h.Guard.WithServingRetirement(ctx, next.Host, hostguard.Captured{Incarnation: next.Incarnation, Stamp: next.Stamp}, func(fresh hostclaim.Observed, locked hostguard.FenceView) error {
86 if locked.Installation != view.Installation {
87 return ErrChanged
88 }
89 if fresh.ETag != next.ETag {
90 return ErrChanged
91 }
92 if err = h.validate(ctx, j, fresh.Claim); err != nil {
93 return err
94 }
95 saved, err := readTrustedJournal(path)
96 if err == nil {
97 var existing coldJournal
98 if decodeStrict(saved, &existing) != nil || !reflect.DeepEqual(existing, j) {
99 return ErrTypedJournal
100 }
101 return nil
102 }
103 if !errors.Is(err, os.ErrNotExist) {
104 return err
105 }
106 return writeColdJournal(path, j)
107 })
108 if err != nil {
109 return err
110 }
111 }
112 check := func(c *hostclaim.Claim) error { return h.validate(ctx, j, c) }
113 if j.Phase == "complete" {
114 return h.Guard.CompleteColdHandover(ctx, j.Barrier, true, check, func() error { return nil })
115 }
116 if err = h.at("intent"); err != nil {
117 return err
118 }
119 if err = h.Guard.BeginServingHandover(ctx, prior, j.Barrier, *j.ServingInstallation, check); err != nil {
120 return err
121 }
122 if err = h.at("barrier"); err != nil {
123 return err
124 }
125 return h.Guard.CompleteColdHandover(ctx, j.Barrier, false, check, func() error {
126 if prior.Session != 0 {
127 if err := h.VerifyReceipt(ctx, j.Receipt); err != nil {
128 return err
129 }
130 }
131 launches, err := h.Launches.RootSessionLaunches(ctx, next.Host, next.Incarnation, next.Stamp, next.Session, next.Box, next.Enrollment)
132 if err != nil {
133 return err
134 }
135 if len(launches) != 0 {
136 return ErrChanged
137 }
138 rebound := j.Inventory
139 rebound.Entries = append([]StopInventoryEntry(nil), j.Inventory.Entries...)
140 rebound.InstallationStamp, rebound.DirectoryRevision = next.Stamp, next.DirectoryRevision
141 for i := range rebound.Entries {
142 e := &rebound.Entries[i]
143 if e.Kind == "path" {
144 a, err := h.directory(ctx, prior, *e)
145 if err != nil {
146 return err
147 }
148 e.Device = a.Dev
149 e.Identity = a.Identity
150 }
151 e.Ownership = j.After.Route.ClaimOwner
152 if e.Kind == "route" {
153 e.RowIdentity = j.After.Route.ClaimRowID
154 }
155 if e.Kind == "timer" {
156 for _, timer := range j.After.Timers {
157 if e.RelativePath == next.Host+"/"+timer.Name {
158 e.RowIdentity = timer.ClaimRowID
159 }
160 }
161 }
162 }
163 existing, err := h.Inventory.ReadInstallation(ctx, next.Host, next.Incarnation, next.Stamp)
164 if err == nil {
165 if !reflect.DeepEqual(existing, rebound) {
166 return ErrTypedInventory
167 }
168 } else if !errors.Is(err, sql.ErrNoRows) {
169 return err
170 } else if err = h.Inventory.WriteStopInventory(ctx, rebound); err != nil {
171 return err
172 }
173 if err = h.at("inventory"); err != nil {
174 return err
175 }
176 if err = h.Registry.RebindClaimRows(ctx, j.Before, j.After); err != nil {
177 return err
178 }
179 if err = h.at("rows"); err != nil {
180 return err
181 }
182 j.Phase = "complete"
183 if err = writeColdJournal(path, j); err != nil {
184 return err
185 }
186 return h.at("complete-journal")
187 })
188}
189
190func (h ServingHandover) validate(ctx context.Context, j coldJournal, c *hostclaim.Claim) error {
191 n, p := j.Successor, j.Prior
192 if c == nil || c.Host != n.Host || c.Holder == nil || c.Op.Kind != hostclaim.KindServing || c.Op.ID != "" || c.Op.Phase != "" || c.Op.Exec != n.Box || c.Holder.Incarnation != n.Incarnation || c.Holder.Stamp != n.Stamp || c.Holder.Session != n.Session || c.Holder.Generation != n.Generation || c.Holder.WriterBox != n.Box {
193 return ErrChanged
194 }
195 if j.ServingInstallation == nil || (p.Session != 0 && (j.Receipt.Target != p || h.VerifyReceipt(ctx, j.Receipt) != nil)) || (p.Session == 0 && !reflect.DeepEqual(j.Receipt, hostguard.FenceReceipt{})) {
196 return ErrTypedJournal
197 }
198 current, rev, err := h.Inventory.Snapshot(ctx, p.Host, hostguard.Captured{Incarnation: p.Incarnation, Stamp: p.Stamp})
199 if err != nil {
200 return err
201 }
202 if rev != j.ServingInstallation.InventoryRevision || !reflect.DeepEqual(current, j.Inventory) {
203 return ErrTypedInventory
204 }
205 if err = ValidateAgentInventory(j.Inventory, &j.Before.Route, h.Roots["data"]); err != nil {
206 return err
207 }
208 stateRelative, err := filepath.Rel(h.Roots["data"], filepath.Dir(provision.NestedDBPath(h.Roots["data"], p.Host)))
209 if err != nil {
210 return err
211 }
212 routes, paths, timers := 0, 0, 0
213 for _, e := range j.Inventory.Entries {
214 if e.Materialization != "materialized" || e.Ownership != j.Before.Route.ClaimOwner {
215 return ErrTypedInventory
216 }
217 switch e.Kind {
218 case "route":
219 routes++
220 case "path":
221 paths++
222 if e.ManagedRoot != "data" || e.RelativePath != stateRelative || e.Type != 0040000 || e.Device == 0 || e.Inode == 0 {
223 return ErrTypedInventory
224 }
225 if _, err := h.directory(ctx, p, e); err != nil {
226 return err
227 }
228 case "timer":
229 timers++
230 found := 0
231 for _, timer := range j.Before.Timers {
232 if e.RelativePath == p.Host+"/"+timer.Name && timer.ClaimOwnership.Valid() && timer.ClaimIncarnation == p.Incarnation && timer.ClaimStamp == p.Stamp && timer.ClaimOwner == e.Ownership && timer.ClaimRowID == e.RowIdentity && timer.Registration != "" {
233 found++
234 }
235 }
236 if found != 1 {
237 return ErrTypedInventory
238 }
239 default:
240 return ErrTypedInventory
241 }
242 }
243 if routes != 1 || paths > 1 || (paths == 0 && p.Session != 0) || timers != len(j.Before.Timers) {
244 return ErrTypedInventory
245 }
246 return h.Validate(ctx, c, p, n, j.Inventory, *j.ServingInstallation)
247}
248
249func (h ServingHandover) directory(ctx context.Context, prior hostguard.FenceTarget, e StopInventoryEntry) (Artifact, error) {
250 a, err := (TypedParker{Roots: h.Roots, ParkedRoots: h.Roots, Incarnation: prior.Incarnation, Host: prior.Host, Retirement: "serving-handover"}).CaptureArtifact(e.ManagedRoot, e.RelativePath, e.Ownership, ArtifactPath)
251 if err != nil {
252 return a, err
253 }
254 if a.Ino != e.Inode || a.Type != e.Type {
255 return a, ErrChanged
256 }
257 if h.VerifyDirectory == nil {
258 return a, ErrChanged
259 }
260 if err := h.VerifyDirectory(ctx, prior, e, a); err != nil {
261 return a, err
262 }
263 return a, nil
264}
265
266func (h ServingHandover) Recover(ctx context.Context) error {
267 entries, err := readTrustedJournalDir(h.JournalDir)
268 if err != nil {
269 return err
270 }
271 if len(entries) == 0 {
272 return nil
273 }
274 pending, err := h.Guard.PendingFences(ctx)
275 if err != nil {
276 return err
277 }
278 var failures []error
279 for _, e := range entries {
280 if e.IsDir() || filepath.Ext(e.Name()) != ".json" {
281 continue
282 }
283 body, err := readTrustedJournal(filepath.Join(h.JournalDir, e.Name()))
284 if err != nil {
285 failures = append(failures, err)
286 continue
287 }
288 var j coldJournal
289 if decodeStrict(body, &j) != nil || j.V != 1 || j.ID+".json" != e.Name() || j.ServingInstallation == nil {
290 failures = append(failures, ErrTypedJournal)
291 continue
292 }
293 if j.Phase == "complete" {
294 found := false
295 for _, target := range pending {
296 found = found || target == j.Barrier
297 }
298 if !found {
299 continue
300 }
301 }
302 if err = h.Run(ctx, j.Prior, j.Successor); err != nil {
303 failures = append(failures, err)
304 continue
305 }
306 }
307 return errors.Join(failures...)
308}
309func (h ServingHandover) at(point string) error {
310 if h.checkpoint != nil {
311 return h.checkpoint(point)
312 }
313 return nil
314}
315