amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package host
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "os"
9 "time"
10
11 "github.com/carlosframework/platform/internal/hostclaim"
12 "github.com/carlosframework/platform/internal/hostguard"
13 "github.com/carlosframework/platform/internal/hostops"
14 "github.com/carlosframework/platform/internal/provision"
15 "github.com/carlosframework/platform/internal/rootparticipation"
16 "github.com/carlosframework/platform/internal/rootstate"
17)
18
19type rootMigrationIdentity struct {
20 Deployment, Box, Enrollment string
21 Generation uint64
22}
23type rootMigrationItem struct {
24 Installation hostguard.RootInstallation
25 Session uint64
26 ClaimETag string
27}
28type rootMigrationSnapshot struct {
29 Complete bool
30 DirectoryRevision uint64
31 Exclusion string
32 Records []provision.Record
33 Claims []hostclaim.Claim
34 Items []rootMigrationItem
35 MoveJournals, MoveReservations int
36 Report hostops.CensusReport
37}
38type rootMigrationRecord struct {
39 Schema int
40 Identity rootMigrationIdentity
41 Phase string
42 Observation string
43 Snapshot rootMigrationSnapshot
44}
45type rootMigration struct {
46 prepareRecovery func(context.Context) error
47 prepareFloors func(context.Context) error
48 guardReady func(context.Context) error
49 state, execution rootstate.Store
50 identity rootMigrationIdentity
51 quiesce func(context.Context) (func(), error)
52 observe func(context.Context) (rootMigrationSnapshot, error)
53 deny, recover func(context.Context, rootMigrationItem) error
54 ready func(context.Context, rootMigrationSnapshot) error
55 publish func(context.Context, hostops.CensusReport) error
56}
57
58func (s rootMigrationSnapshot) preflight(id rootMigrationIdentity) error {
59 if s.Exclusion == "" || s.DirectoryRevision == 0 || s.MoveJournals != 0 || s.MoveReservations != 0 || s.Items == nil || !s.Report.Complete || s.Report.Box != id.Box || s.Report.Enrollment != id.Enrollment || s.Report.DirectoryRevision != s.DirectoryRevision || s.Report.Reservation == "" || s.Report.BootID == "" || s.Report.CapturedAt.IsZero() || s.Report.Installations == nil || s.Report.Seeds == nil || len(s.Report.Seeds.Journaled) > 0 || len(s.Report.Seeds.Gated) > 0 {
60 return errRootMigrationBusy
61 }
62 for _, rec := range s.Records {
63 if !claimNamedUnit4aComplete && rec.Unit != "" {
64 for _, item := range s.Items {
65 if item.Installation.Host == rec.Host {
66 return errRootMigrationBusy
67 }
68 }
69 }
70 if rec.Seed.Unfinished() {
71 return errRootMigrationBusy
72 }
73 }
74 seen := map[string]bool{}
75 for _, item := range s.Items {
76 in := item.Installation
77 if in.Host == "" || seen[fmt.Sprintf("%s/%s/%d/%d", in.Host, in.Incarnation, in.Stamp, item.Session)] || in.Incarnation == "" || in.Stamp == 0 || in.Account == "" || in.App == "" || in.Record == "" || in.Box != id.Box || in.Enrollment != id.Enrollment || in.Migration != id.Generation || item.Session == 0 || item.ClaimETag == "" {
78 return errRootMigrationBusy
79 }
80 seen[fmt.Sprintf("%s/%s/%d/%d", in.Host, in.Incarnation, in.Stamp, item.Session)] = true
81 }
82 return rootMigrationPreflight(s.Complete, s.Records, s.Claims)
83}
84func (s rootMigrationSnapshot) fingerprint() string {
85 // Wall time is report metadata, not census membership or exclusion identity.
86 s.Report.CapturedAt = time.Time{}
87 s.Report.BootID = ""
88 s.Items = append([]rootMigrationItem(nil), s.Items...)
89 for i := range s.Items {
90 s.Items[i].Installation.InventoryRevision = 0
91 s.Items[i].Installation.Artifacts = hostguard.InstallationArtifacts{}
92 }
93 return generationInventoryDigest(s)
94}
95func (m rootMigration) write(name string, v any) error {
96 raw, err := json.Marshal(v)
97 if err != nil {
98 return err
99 }
100 return m.state.Write(name, raw)
101}
102func (m rootMigration) census(complete bool) error {
103 c := rootparticipation.Census{Schema: 1, Box: m.identity.Box, Enrollment: m.identity.Enrollment, Migration: m.identity.Generation, Revision: 1, LaunchRevision: 1, Complete: complete}
104 if !complete {
105 c.Pending = "migration"
106 }
107 return (rootparticipation.Ledger{State: m.execution}).MigrationCensus(context.Background(), c, false)
108}
109func (m rootMigration) run(ctx context.Context) error {
110 if m.state == nil || m.execution == nil || m.identity.Deployment == "" || m.identity.Box == "" || m.identity.Enrollment == "" || m.identity.Generation == 0 || m.quiesce == nil || m.observe == nil || m.deny == nil || m.recover == nil || m.ready == nil || m.publish == nil {
111 return errRootMigrationBusy
112 }
113 unlock, err := m.state.Lock(ctx, "migration.lock")
114 if err != nil {
115 return err
116 }
117 defer unlock()
118 var saved rootMigrationRecord
119 raw, err := m.state.Read("migration.json")
120 exists := err == nil
121 if exists {
122 if rootparticipation.Decode(raw, &saved) != nil || saved.Schema != 1 || saved.Identity != m.identity || saved.Snapshot.preflight(m.identity) != nil {
123 return errRootMigrationBusy
124 }
125 switch saved.Phase {
126 case "prepared", "fenced", "recovered", "publishing", "complete":
127 default:
128 return errRootMigrationBusy
129 }
130 } else if !errors.Is(err, os.ErrNotExist) {
131 return err
132 }
133 release, err := m.quiesce(ctx)
134 if err != nil {
135 return err
136 }
137 defer release()
138 if exists && saved.Phase == "complete" {
139 return (rootparticipation.Ledger{State: m.execution}).MigrationCensus(ctx, rootparticipation.Census{Box: m.identity.Box, Enrollment: m.identity.Enrollment, Migration: m.identity.Generation}, true)
140 }
141 if m.prepareFloors == nil || m.guardReady == nil {
142 return errRootMigrationBusy
143 }
144 // The floor directories are repaired under the lock and quiesce, before
145 // guardReady's first floor read: a directory an earlier attempt left
146 // 0700 would otherwise refuse that read on every rerun.
147 if err := m.prepareFloors(ctx); err != nil {
148 return err
149 }
150 if err := m.guardReady(ctx); err != nil {
151 return err
152 }
153 if exists && saved.Phase == "publishing" {
154 if err := m.census(true); err != nil {
155 return err
156 }
157 if err := m.publish(ctx, saved.Snapshot.Report); err != nil {
158 return err
159 }
160 saved.Phase = "complete"
161 return m.write("migration.json", saved)
162 }
163 observe := func() (rootMigrationSnapshot, error) {
164 s, e := m.observe(ctx)
165 if e == nil {
166 e = s.preflight(m.identity)
167 }
168 return s, e
169 }
170 first, err := observe()
171 if err != nil {
172 return err
173 }
174 fresh, err := observe()
175 if err != nil {
176 return err
177 }
178 if first.fingerprint() != fresh.fingerprint() {
179 return errRootMigrationBusy
180 }
181 if !exists {
182 saved = rootMigrationRecord{Schema: 1, Identity: m.identity}
183 }
184 // Keep every old tuple while updating current claim evidence. A second
185 // observation above proves the new snapshot under the same exclusion.
186 items := append([]rootMigrationItem(nil), fresh.Items...)
187 for _, old := range saved.Snapshot.Items {
188 found := false
189 for _, item := range items {
190 if old.Installation.Host == item.Installation.Host && old.Installation.Incarnation == item.Installation.Incarnation && old.Installation.Stamp == item.Installation.Stamp && old.Session == item.Session {
191 found = true
192 break
193 }
194 }
195 if !found {
196 items = append(items, old)
197 }
198 }
199 saved.Observation = fresh.fingerprint()
200 saved.Snapshot = fresh
201 saved.Snapshot.Items = items
202 if saved.Snapshot.Items == nil {
203 saved.Snapshot.Items = []rootMigrationItem{}
204 }
205
206 if err := m.census(false); err != nil {
207 return err
208 }
209 saved.Phase = "prepared"
210 if err := m.write("migration.json", saved); err != nil {
211 return err
212 }
213 for _, item := range saved.Snapshot.Items {
214 if err := m.deny(ctx, item); err != nil {
215 return err
216 }
217 if err := m.write("deny-"+generationInventoryDigest([]any{item.Installation.Host, item.Installation.Incarnation, item.Installation.Stamp, item.Session})+".json", item); err != nil {
218 return err
219 }
220 }
221 saved.Phase = "fenced"
222 if err := m.write("migration.json", saved); err != nil {
223 return err
224 }
225 if m.prepareRecovery != nil {
226 if err := m.prepareRecovery(ctx); err != nil {
227 return err
228 }
229 }
230 for _, item := range saved.Snapshot.Items {
231 // Reverify even a visible receipt on retry: no record can manufacture stop
232 // evidence, and the recovery adapter preserves the original signed time.
233 if err := m.recover(ctx, item); err != nil {
234 return err
235 }
236 if err := m.write("receipt-"+generationInventoryDigest([]any{item.Installation.Host, item.Installation.Incarnation, item.Installation.Stamp, item.Session})+".json", item); err != nil {
237 return err
238 }
239 }
240 saved.Phase = "recovered"
241 if err := m.write("migration.json", saved); err != nil {
242 return err
243 }
244 fresh, err = observe()
245 if err != nil {
246 return err
247 }
248 if saved.Observation != fresh.fingerprint() {
249 return errRootMigrationBusy
250 }
251 if err := m.ready(ctx, saved.Snapshot); err != nil {
252 return err
253 }
254 if err := m.write("report.json", saved.Snapshot.Report); err != nil {
255 return err
256 }
257 saved.Phase = "publishing"
258 if err := m.write("migration.json", saved); err != nil {
259 return err
260 }
261 if err := m.census(true); err != nil {
262 return err
263 }
264 if err := m.publish(ctx, saved.Snapshot.Report); err != nil {
265 return err
266 }
267 saved.Phase = "complete"
268 return m.write("migration.json", saved)
269}
270