amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package hostclaim
2
3import (
4 "bytes"
5 "context"
6 "crypto/ed25519"
7 "encoding/json"
8 "errors"
9 "fmt"
10 "io"
11 "reflect"
12 "strings"
13 "time"
14
15 "github.com/carlosframework/platform/internal/release"
16)
17
18var (
19 ErrLossOwnerRequired = errors.New("hostclaim: operation owner must authorize lost-box transition")
20 ErrLossAlreadyApplied = errors.New("hostclaim: loss transition already applied")
21 ErrLossEvidencePending = errors.New("hostclaim: admission loss evidence is not archived")
22)
23
24// LostDeclaration is the exact claim observation and approved effect set from
25// one prepared durable loss record. ExpectedClaimETag is checked before Update
26// and the generation is checked again inside its transition.
27type LostDeclaration struct {
28 LossID string
29 ExpectedClaimGen uint64
30 ExpectedClaimETag string
31 HolderIncarnation string
32 HolderStamp uint64
33 HolderSession uint64
34 OpID string
35 OpKind string
36 OpPhase string
37 Box string
38 QuarantineRevision uint64
39 Enrollment string
40 FencedAt time.Time
41 Consequences LossConsequences
42}
43
44// LossOperationOwner owns the legal transition for its operation kind. A
45// generic declare-lost path cannot infer moves' source/target or record-commit
46// state, so a missing owner always refuses.
47type LossOperationOwner interface {
48 ValidateDeclareLost(Claim, LostDeclaration) error
49}
50
51// DeclareLost reads and verifies the root-signed prepared declaration, then
52// conditionally commits its exact effects. Owning both reads and Update here
53// makes the durable record, claim ETag and claim generation enforceable API
54// inputs rather than caller conventions.
55func DeclareLost(ctx context.Context, s release.Store, pub ed25519.PublicKey, observed Observed, lossID string, owner LossOperationOwner) (Observed, error) {
56 if owner == nil {
57 return Observed{}, ErrLossOwnerRequired
58 }
59 loss, err := ReadLossRecord(ctx, s, pub, lossID)
60 if err != nil {
61 return Observed{}, err
62 }
63 in := declarationFromRecord(loss.Record)
64 if observed.Absent || observed.Claim == nil || observed.ETag == "" || observed.Host != observed.Claim.Host || observed.Host != loss.Record.Host {
65 return Observed{}, ErrConflict
66 }
67 if lossRefPresent(observed.Claim, in.LossID) {
68 // loss_refs is the permanent commit marker. Readmission intentionally
69 // removes Lost/LostRevisions while preserving this evidence.
70 return observed, nil
71 }
72 if loss.Record.State != "prepared" || observed.ETag != in.ExpectedClaimETag || observed.Claim.Gen != in.ExpectedClaimGen {
73 return Observed{}, ErrConflict
74 }
75 return Update(ctx, s, observed, declareLostTransition(in, owner))
76}
77
78func declarationFromRecord(r LossRecord) LostDeclaration {
79 return LostDeclaration{
80 LossID: r.ID, ExpectedClaimGen: r.ExpectedClaimGen, ExpectedClaimETag: r.ExpectedClaimETag,
81 HolderIncarnation: r.HolderIncarnation, HolderStamp: r.HolderStamp, HolderSession: r.HolderSession,
82 OpID: r.OpID, OpKind: r.OpKind, OpPhase: r.OpPhase, Box: r.Box, Enrollment: r.Enrollment,
83 QuarantineRevision: r.QuarantineRevision, FencedAt: r.FencedAt, Consequences: r.ConfirmedConsequences,
84 }
85}
86
87// AdmitWriterWithLoss binds every target and source-cut tuple to the
88// root-signed durable declaration before the claim CAS. Direct AdmitWriter
89// accepts native additional losses but refuses an unverified LossID.
90type ValidatedWriteSource struct {
91 Reason string
92 Box string
93 Generation string
94 Session uint64
95 Region string
96 Baseline string
97 RemoteMaxTXID uint64
98}
99
100// LossRecoveryOwner validates the operation's persisted recovery handover
101// against the original declaration and the requested admission. It must check
102// the legal kind/phase transition, recovery intent, executor and installation
103// stamp. The library retains incarnation, operation and frozen-loss bindings;
104// it cannot infer the moves owner's legal recovery path. Inputs are copies.
105type LossRecoveryOwner interface {
106 ValidateLossRecovery(Claim, LossRecord, Admission) error
107}
108
109func AdmitWriterWithLoss(ctx context.Context, s release.Store, pub ed25519.PublicKey, observed Observed, in Admission, lossID string, validated []ValidatedWriteSource, owner LossRecoveryOwner) (Observed, error) {
110 if owner == nil {
111 return Observed{}, ErrLossOwnerRequired
112 }
113 if observed.Absent || observed.Claim == nil || observed.Claim.Holder == nil || !lossRefPresent(observed.Claim, lossID) {
114 return Observed{}, ErrLossRecord
115 }
116 loss, err := ReadLossRecord(ctx, s, pub, lossID)
117 if err != nil {
118 return Observed{}, err
119 }
120 if loss.Record.State != "committed" || loss.Record.Host != observed.Host || loss.Record.HolderIncarnation != observed.Claim.Holder.Incarnation ||
121 loss.Record.OpID == "" || loss.Record.OpID != observed.Claim.Op.ID {
122 return Observed{}, ErrLossRecord
123 }
124 if in.Mode != "cold" || in.PredecessorCertified || in.Coverage != "" {
125 return Observed{}, fmt.Errorf("%w: a declared target loss requires an uncovered cold admission", ErrLossRecord)
126 }
127 h := observed.Claim.Holder
128 predecessor := AdmissionRecord{Session: h.Session, Box: h.WriterBox, Generation: h.Generation, Region: in.PredecessorRegion}
129 if len(h.Admissions) > 0 {
130 predecessor = h.Admissions[len(h.Admissions)-1]
131 }
132 for i, prior := range h.Admissions {
133 if prior.LossID != lossID {
134 continue
135 }
136 predecessorRegion := in.PredecessorRegion
137 if i > 0 {
138 predecessorRegion = h.Admissions[i-1].Region
139 }
140 if prior.Box != in.Box || prior.Mode != in.Mode || prior.Generation != in.NewGeneration ||
141 prior.Region != in.Region || prior.PredecessorRegion != predecessorRegion || prior.Coverage != in.Coverage ||
142 prior.Predecessor != in.ObservedGeneration || prior.WritesLost == nil ||
143 prior.WritesLost.Generation != in.ObservedGeneration || prior.WritesLost.Session != in.ObservedSession ||
144 prior.WritesLost.RemoteMaxTXID != in.RemoteMaxTXID || !reflect.DeepEqual(prior.AdditionalWritesLost, in.AdditionalWritesLost) {
145 return Observed{}, fmt.Errorf("%w: declaration id already used by another admission", ErrWrongState)
146 }
147 predecessor = AdmissionRecord{Box: prior.WritesLost.Box, Generation: prior.WritesLost.Generation, Session: prior.WritesLost.Session, Region: prior.PredecessorRegion}
148 break
149 }
150 want := map[string]WriteSource{}
151 for _, source := range loss.Record.WriteSources {
152 key := fmt.Sprintf("%s\x00%s\x00%d", source.Box, source.Generation, source.Session)
153 want[key] = source
154 }
155 targetKey := fmt.Sprintf("%s\x00%s\x00%d", predecessor.Box, predecessor.Generation, predecessor.Session)
156 if len(want) != 0 && want[targetKey].Reason != "target" && want[targetKey].Reason != "source-cut" {
157 return Observed{}, fmt.Errorf("%w: predecessor is not the declared lost target", ErrLossRecord)
158 }
159 primary := WritesLost{Box: predecessor.Box, Generation: predecessor.Generation, Session: predecessor.Session, RemoteMaxTXID: in.RemoteMaxTXID}
160 if len(want) != 0 && !validWriteLosses(&primary, in.AdditionalWritesLost) {
161 return Observed{}, ErrLossRecord
162 }
163 declared := map[string]WritesLost{}
164 if len(want) != 0 {
165 declared[targetKey] = primary
166 }
167 for _, additional := range in.AdditionalWritesLost {
168 key := fmt.Sprintf("%s\x00%s\x00%d", additional.Box, additional.Generation, additional.Session)
169 if _, ok := want[key]; !ok {
170 return Observed{}, fmt.Errorf("%w: additional tuple was not frozen by the declaration", ErrLossRecord)
171 }
172 declared[key] = additional
173 }
174 if len(declared) != len(want) {
175 return Observed{}, fmt.Errorf("%w: admission omitted a frozen write-loss tuple", ErrLossRecord)
176 }
177 provided := map[string]ValidatedWriteSource{}
178 for _, source := range validated {
179 key := fmt.Sprintf("%s\x00%s\x00%d", source.Box, source.Generation, source.Session)
180 if _, duplicate := provided[key]; duplicate {
181 return Observed{}, ErrLossRecord
182 }
183 provided[key] = source
184 }
185 if len(provided) != len(want) {
186 return Observed{}, fmt.Errorf("%w: admission omitted validated baseline evidence", ErrLossRecord)
187 }
188 for key, frozen := range want {
189 actual, ok := provided[key]
190 if !ok || actual.Reason != frozen.Reason || actual.Region != frozen.Region || actual.Baseline != frozen.Baseline {
191 return Observed{}, fmt.Errorf("%w: validated baseline differs from frozen source", ErrLossRecord)
192 }
193 declaration, ok := declared[key]
194 if !ok || declaration.RemoteMaxTXID != actual.RemoteMaxTXID {
195 return Observed{}, fmt.Errorf("%w: declared remote maximum differs from its validated baseline", ErrLossRecord)
196 }
197 }
198 // Owner policy may inspect detached inputs but cannot alter the claim,
199 // frozen declaration or the admission that passed the evidence checks.
200 policy := struct {
201 Claim Claim
202 Loss LossRecord
203 Admission Admission
204 }{*observed.Claim, loss.Record, in}
205 wire, err := json.Marshal(policy)
206 if err != nil {
207 return Observed{}, err
208 }
209 // Clear shared maps, slices and pointers before decoding the copies.
210 policy.Claim, policy.Loss, policy.Admission = Claim{}, LossRecord{}, Admission{}
211 if err := json.Unmarshal(wire, &policy); err != nil {
212 return Observed{}, err
213 }
214 if err := owner.ValidateLossRecovery(policy.Claim, policy.Loss, policy.Admission); err != nil {
215 return Observed{}, err
216 }
217 in.LossID, in.lossVerified = lossID, true
218 next, err := Update(ctx, s, observed, AdmitWriter(in))
219 if errors.Is(err, ErrLossAlreadyApplied) {
220 return observed, nil
221 }
222 return next, err
223}
224
225func declareLostTransition(in LostDeclaration, owner LossOperationOwner) func(*Claim) error {
226 return func(c *Claim) error {
227 if owner == nil {
228 return ErrLossOwnerRequired
229 }
230 if !safeLossID(in.LossID) || in.ExpectedClaimGen == 0 || in.ExpectedClaimETag == "" || in.QuarantineRevision == 0 ||
231 in.Box == "" || in.Enrollment == "" || in.FencedAt.IsZero() || !validConsequences(in.Consequences) {
232 return ErrLossRecord
233 }
234 if c.Gen != in.ExpectedClaimGen || c.Holder == nil || c.Holder.Incarnation != in.HolderIncarnation ||
235 c.Holder.Stamp != in.HolderStamp || c.Holder.Session != in.HolderSession || c.Op.ID != in.OpID ||
236 c.Op.Kind != in.OpKind || c.Op.Phase != in.OpPhase {
237 return fmt.Errorf("%w: loss declaration no longer matches the holder and operation", ErrStaleOp)
238 }
239 if old := c.LostRevisions[in.Box]; old != 0 && in.QuarantineRevision <= old {
240 return fmt.Errorf("%w: declaration requires a strictly newer quarantine revision", ErrWrongState)
241 }
242 if lossRefPresent(c, in.LossID) {
243 return ErrLossAlreadyApplied
244 }
245 for key := range c.Op.Journal {
246 if strings.HasPrefix(key, "loss.") && !(destroyedLocalOnly(in.Consequences) && stringPresent(c.Lost, in.Box)) {
247 return ErrWrongState
248 }
249 }
250 copy, err := cloneClaim(c)
251 if err != nil {
252 return err
253 }
254 // The policy callback may inspect copies, never alter the signed effects.
255 ownerBytes, err := json.Marshal(in)
256 if err != nil {
257 return err
258 }
259 var ownerInput LostDeclaration
260 if err := json.Unmarshal(ownerBytes, &ownerInput); err != nil {
261 return err
262 }
263 if err := owner.ValidateDeclareLost(*copy, ownerInput); err != nil {
264 return err
265 }
266
267 if c.Op.Kind == KindServing || destroyedLocalOnly(in.Consequences) {
268 if c.Op.Kind != KindServing && c.Op.Exec == in.Box {
269 return ErrWrongState
270 }
271 if (c.Op.Kind == KindServing && (in.OpID != "" || in.OpPhase != "")) || !destroyedLocalOnly(in.Consequences) {
272 return ErrLossRecord
273 }
274 // No operation journal, authority, regional reassignment or holder
275 // mutation is permitted in this branch. Original chore identities
276 // still drive discovery completion after adoption cleared op.id.
277 if err := dischargeDestroyedLocal(c, in); err != nil {
278 return err
279 }
280 recordLost(c, in)
281 return nil
282 }
283 if in.OpID == "" {
284 return ErrLossRecord
285 }
286
287 isExec := c.Op.Exec == in.Box
288 isParticipant := stringPresent(c.Op.Participants, in.Box)
289 if isExec != in.Consequences.RevokeExecution || isParticipant != in.Consequences.RemoveParticipation {
290 return fmt.Errorf("%w: confirmed authority consequences do not match the claim", ErrWrongState)
291 }
292 stamp, session := c.Holder.Stamp, c.Holder.Session
293 intent, err := canonicalRaw(in.Consequences.RecoveryIntent)
294 if err != nil {
295 return ErrLossRecord
296 }
297 if c.Op.Journal == nil {
298 c.Op.Journal = map[string]string{}
299 }
300 journalKey := "loss." + in.LossID
301 if prior, exists := c.Op.Journal[journalKey]; exists && prior != intent {
302 return fmt.Errorf("%w: declaration recovery intent changed", ErrWrongState)
303 }
304 // Recovery intent is recorded before any authority or chore effects in
305 // this transition. Update makes the entire sequence one claim CAS.
306 c.Op.Journal[journalKey] = intent
307 gate, err := json.Marshal(operationLossGate{Box: in.Box, RevokeExecution: in.Consequences.RevokeExecution})
308 if err != nil {
309 return err
310 }
311 c.Op.Journal["loss_gate."+in.LossID] = string(gate)
312
313 if isExec {
314 c.Op.Exec = "none"
315 }
316 if isParticipant {
317 c.Op.Participants = removeString(c.Op.Participants, in.Box)
318 }
319 ownedRegional := map[string]bool{}
320 for _, o := range c.Obligations {
321 if (o.Kind == "park-replica" || o.Kind == "park-generation") && o.Owner == "box:"+in.Box {
322 ownedRegional[o.Op+"\x00"+o.ID+"\x00"+o.Retirement] = true
323 }
324 }
325 confirmedRegional := map[string]bool{}
326 for _, ref := range in.Consequences.ReassignRegional {
327 confirmedRegional[ref.Op+"\x00"+ref.ID+"\x00"+ref.Retirement] = true
328 }
329 if !reflect.DeepEqual(ownedRegional, confirmedRegional) {
330 return fmt.Errorf("%w: confirmed regional reassignments do not exactly cover lost-box work", ErrWrongState)
331 }
332 for _, ref := range in.Consequences.ReassignRegional {
333 o, ok := exactObligation(c, ref)
334 if !ok || (o.Kind != "park-replica" && o.Kind != "park-generation") || o.Region == "" || o.Owner != "box:"+in.Box {
335 return fmt.Errorf("%w: regional obligation %q changed", ErrWrongState, ref.ID)
336 }
337 o.Owner = "region:" + o.Region
338 }
339 if err := dischargeDestroyedLocal(c, in); err != nil {
340 return err
341 }
342 if c.Holder.Stamp != stamp || c.Holder.Session != session {
343 return fmt.Errorf("%w: declare-lost changed installation authority", ErrMalformed)
344 }
345 recordLost(c, in)
346 return nil
347 }
348}
349
350func recordLost(c *Claim, in LostDeclaration) {
351 if !stringPresent(c.Lost, in.Box) {
352 c.Lost = append(c.Lost, in.Box)
353 }
354 if c.LostRevisions == nil {
355 c.LostRevisions = map[string]uint64{}
356 }
357 c.LostRevisions[in.Box] = in.QuarantineRevision
358 c.LossRefs = append(c.LossRefs, in.LossID)
359}
360
361func dischargeDestroyedLocal(c *Claim, in LostDeclaration) error {
362 for _, ref := range in.Consequences.DischargeLocal {
363 o, ok := exactObligation(c, ref)
364 if !ok || o.Kind != "park-local" || o.Box != in.Box || !destroyedStorageCovers(in, ref) {
365 return fmt.Errorf("%w: local obligation %q lacks exact destroyed-storage evidence", ErrWrongState, ref.ID)
366 }
367 if err := CompleteObligation(ref.Op, ref.ID)(c); err != nil {
368 return err
369 }
370 }
371 return nil
372}
373
374func destroyedStorageCovers(in LostDeclaration, ref ObligationRef) bool {
375 needle := ref.Op + "/" + ref.ID + "/" + ref.Retirement
376 for _, covered := range in.Consequences.DestroyedStorage {
377 if covered == needle {
378 return true
379 }
380 }
381 return false
382}
383
384func exactObligation(c *Claim, ref ObligationRef) (*Obligation, bool) {
385 for i := range c.Obligations {
386 o := &c.Obligations[i]
387 if o.ID == ref.ID && o.Op == ref.Op && o.Retirement == ref.Retirement {
388 return o, true
389 }
390 }
391 return nil, false
392}
393
394func cloneClaim(c *Claim) (*Claim, error) {
395 raw, err := json.Marshal(c)
396 if err != nil {
397 return nil, err
398 }
399 return decodeStrict(raw)
400}
401
402func canonicalRaw(raw json.RawMessage) (string, error) {
403 var value any
404 decoder := json.NewDecoder(bytes.NewReader(raw))
405 decoder.UseNumber()
406 if err := decoder.Decode(&value); err != nil {
407 return "", err
408 }
409 if err := decoder.Decode(new(any)); err != io.EOF {
410 return "", ErrLossRecord
411 }
412 canonical, err := json.Marshal(value)
413 return string(canonical), err
414}
415
416func lossRefPresent(c *Claim, id string) bool { return stringPresent(c.LossRefs, id) }
417
418func stringPresent(values []string, wanted string) bool {
419 for _, value := range values {
420 if value == wanted {
421 return true
422 }
423 }
424 return false
425}
426
427func removeString(values []string, unwanted string) []string {
428 out := values[:0]
429 for _, value := range values {
430 if value != unwanted {
431 out = append(out, value)
432 }
433 }
434 return out
435}
436
437func validRegionalOwner(owner, region string) bool {
438 if owner == "region:"+region {
439 return true
440 }
441 return len(owner) > len("box:") && owner[:len("box:")] == "box:" && validBoxReference(owner[len("box:"):])
442}
443
444// Readmit removes only the quarantine entry for the exact revision being
445// cleared. Durable declaration references intentionally survive.
446func Readmit(box string, quarantineRevision uint64) func(*Claim) error {
447 return func(c *Claim) error {
448 if box == "" || quarantineRevision == 0 || c.LostRevisions[box] != quarantineRevision || !stringPresent(c.Lost, box) {
449 return fmt.Errorf("%w: quarantine revision does not match", ErrWrongState)
450 }
451 c.Lost = removeString(c.Lost, box)
452 delete(c.LostRevisions, box)
453 if len(c.LostRevisions) == 0 {
454 c.LostRevisions = nil
455 }
456 return nil
457 }
458}
459
460func admissionHasLoss(a AdmissionRecord) bool {
461 return a.WritesLost != nil || len(a.AdditionalWritesLost) != 0
462}
463
464func admissionLossArchived(a AdmissionRecord) bool {
465 return !admissionHasLoss(a) || a.LossArchivedClaimGen == a.Session
466}
467
468// RequireLossEvidenceArchived is called by every holder-removal transition.
469func RequireLossEvidenceArchived(c *Claim) error {
470 if c.Holder == nil {
471 return nil
472 }
473 for _, a := range c.Holder.Admissions {
474 if !admissionLossArchived(a) {
475 return ErrLossEvidencePending
476 }
477 }
478 return nil
479}
480
481// ArchiveAdmissionLosses reads the root-signed committed record and marks only
482// entries whose exact declarations and admission claim generation are already
483// durable there. The pure transition is private so callers cannot fabricate
484// archive proof.
485func ArchiveAdmissionLosses(ctx context.Context, s release.Store, pub ed25519.PublicKey, observed Observed, lossID string) (Observed, error) {
486 loss, err := ReadLossRecord(ctx, s, pub, lossID)
487 if err != nil {
488 return Observed{}, err
489 }
490 if loss.Record.State != "committed" || loss.Record.Host != observed.Host {
491 return Observed{}, ErrLossEvidencePending
492 }
493 return Update(ctx, s, observed, archiveAdmissionLossesTransition(lossID, loss.Record.ArchivedAdmissionLosses))
494}
495
496func archiveAdmissionLossesTransition(lossID string, archived []ArchivedAdmissionLoss) func(*Claim) error {
497 return func(c *Claim) error {
498 if c.Holder == nil || !lossRefPresent(c, lossID) {
499 return ErrLossEvidencePending
500 }
501 byGen := map[uint64]ArchivedAdmissionLoss{}
502 for _, a := range archived {
503 if a.LossID != lossID || a.AdmissionClaimGen == 0 || !validWriteLosses(a.WritesLost, a.AdditionalWritesLost) {
504 return ErrLossEvidencePending
505 }
506 byGen[a.AdmissionClaimGen] = a
507 }
508 matched := false
509 for i := range c.Holder.Admissions {
510 a := &c.Holder.Admissions[i]
511 if a.LossID != lossID || !admissionHasLoss(*a) {
512 continue
513 }
514 copy, ok := byGen[a.Session]
515 if !ok || !reflect.DeepEqual(copy.WritesLost, a.WritesLost) || !reflect.DeepEqual(copy.AdditionalWritesLost, a.AdditionalWritesLost) {
516 return ErrLossEvidencePending
517 }
518 a.LossArchivedClaimGen = a.Session
519 matched = true
520 }
521 if !matched {
522 return ErrLossEvidencePending
523 }
524 return nil
525 }
526}
527