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 "errors"
6 "fmt"
7 "github.com/carlosframework/platform/internal/moves"
8 "os"
9 "path/filepath"
10 "strconv"
11
12 "github.com/carlosframework/platform/internal/hostbox"
13 "github.com/carlosframework/platform/internal/hostclaim"
14 "github.com/carlosframework/platform/internal/hostguard"
15 "github.com/carlosframework/platform/internal/park"
16)
17
18// claimArtifactRetirement parks a superseded database and marker under the
19// admitted installation's descriptor guard. Landing probes, restoring and
20// reverting share the same exact inventory checks. Initial source recovery
21// uses the source cut's journal and report; later sessions require root's
22// completed writer-stop receipt. Files and environment paths are unaffected.
23// Continue is bound only by the moves C1 integration; nil refuses.
24type claimArtifactRetirement struct {
25 Local park.Local
26 Continue func(kind, from, to string) bool
27}
28
29func (r claimArtifactRetirement) Retire(ctx context.Context, admitted hostclaim.Observed, in ClaimLaunchInputs, generation string) error {
30 box := r.Local.Box
31 if r.Continue == nil || r.Local.Claims == nil || box == "" || !filepath.IsAbs(r.Local.DataDir) || admitted.Claim == nil || admitted.Claim.Holder == nil || !claimArtifactKind(admitted.Claim.Op.Kind) {
32 return hostbox.ErrChanged
33 }
34 h := admitted.Claim.Holder
35 if generation == "" || generation == h.Generation || !filepath.IsAbs(in.DBPath) {
36 return hostbox.ErrChanged
37 }
38 marker, err := filepath.Rel(r.Local.DataDir, in.DBPath+".gen")
39 if err != nil || !filepath.IsLocal(marker) {
40 return hostbox.ErrChanged
41 }
42 database, err := filepath.Rel(r.Local.DataDir, in.DBPath)
43 if err != nil || !filepath.IsLocal(database) {
44 return hostbox.ErrChanged
45 }
46 marker, database = filepath.ToSlash(marker), filepath.ToSlash(database)
47 fresh, err := hostclaim.Read(ctx, r.Local.Claims, admitted.Host)
48 if err != nil {
49 return err
50 }
51 if !claimSameWriter(admitted, fresh, r.Continue) {
52 return hostbox.ErrChanged
53 }
54 c := fresh.Claim
55 if _, ok := claimArtifactWriter(c, box); !ok {
56 return hostbox.ErrChanged
57 }
58 // The superseded database is this op's: its prepared generation, named by
59 // the op id, or a generation an earlier session on this box was admitted
60 // into under it.
61 known := c.Op.Kind == hostclaim.KindLanding && generation == c.Op.ID
62 if claimRecoveryKind(c.Op.Kind) && generation == c.Op.Journal[hostclaim.JournalSrcGen] {
63 known = true
64 }
65 for _, a := range c.Holder.Admissions {
66 if a.Generation == generation && a.Box == box && a.Session < c.Holder.Session {
67 known = true
68 }
69 }
70 if !known {
71 return hostbox.ErrChanged
72 }
73 paths := c.Inventory[box].Paths
74 stamp := paths[marker]
75 if stamp == 0 || stamp >= c.Holder.Stamp || paths[database] != stamp {
76 return hostbox.ErrChanged
77 }
78 if err := r.artifactsStopped(ctx, admitted.Host, c, box, stamp, generation); err != nil {
79 return err
80 }
81 check := func(now *hostclaim.Claim) error {
82 if !claimSameWriter(fresh, hostclaim.Observed{Host: admitted.Host, Claim: now}, r.Continue) {
83 return hostbox.ErrChanged
84 }
85 if _, ok := claimArtifactWriter(now, box); !ok {
86 return hostbox.ErrChanged
87 }
88 return nil
89 }
90 retirement := fmt.Sprintf("superseded-%d-%d", c.Holder.Session, stamp)
91 before, after, err := r.Local.RetireSuperseded(ctx, admitted.Host, hostguard.Captured{Incarnation: h.Incarnation, Stamp: h.Stamp}, c.Op.ID, retirement, stamp, []string{database, marker}, check)
92 if err != nil {
93 return err
94 }
95 return in.observeClaimWrite(before, after)
96}
97
98// stoppedAt requires root's completed full stop of the writer session that ran
99// installation stamp on box under the op: every writer-session receipt the
100// guard journalled for that installation names one session, verifies byte for
101// byte, and that session was admitted on box before the current one. At most
102// one session runs a stamp: a cold admission takes its own stamp unless it is
103// the first writer since the stamp was issued. No receipt refuses.
104func (r claimArtifactRetirement) stoppedAt(ctx context.Context, host string, c *hostclaim.Claim, box string, stamp uint64) error {
105 receipts, err := r.Local.Guard.CompletedWriterStops(ctx, host, hostguard.Captured{Incarnation: c.Holder.Incarnation, Stamp: stamp}, c.Op.ID, box)
106 if err != nil {
107 return err
108 }
109 if len(receipts) == 0 {
110 return hostguard.ErrNoFenceReceipt
111 }
112 session := receipts[0].Target.Session
113 for _, receipt := range receipts {
114 if receipt.Target.Session != session || verifyRetirementReceipt(ctx, receipt) != nil {
115 return ErrClaimStopReceipt
116 }
117 }
118 for _, a := range c.Holder.Admissions {
119 if a.Session == session && a.Box == box && session < c.Holder.Session {
120 return nil
121 }
122 }
123 return hostbox.ErrChanged
124}
125
126// Recover re-drives retirement before any restore/materialization when a
127// failed inventory CAS left both database and marker parked. The absent
128// marker cannot be the only retry trigger: ownership must be recorded before
129// the successor creates new bytes at those paths. Current tags mean the
130// completed retirement already recorded that ownership.
131func (r claimArtifactRetirement) Recover(ctx context.Context, admitted hostclaim.Observed, in ClaimLaunchInputs) error {
132 if r.Continue == nil || r.Local.Claims == nil || admitted.Claim == nil || admitted.Claim.Holder == nil || !filepath.IsAbs(in.DBPath) {
133 return hostbox.ErrChanged
134 }
135 if !claimArtifactKind(admitted.Claim.Op.Kind) {
136 return nil
137 }
138 if _, err := os.Lstat(in.DBPath + ".gen"); err == nil {
139 return nil
140 } else if !errors.Is(err, os.ErrNotExist) {
141 return err
142 }
143 fresh, err := hostclaim.Read(ctx, r.Local.Claims, admitted.Host)
144 if err != nil {
145 return err
146 }
147 if !claimSameWriter(admitted, fresh, r.Continue) {
148 return hostbox.ErrChanged
149 }
150 c, box := fresh.Claim, r.Local.Box
151 if _, ok := claimArtifactWriter(c, box); !ok {
152 return hostbox.ErrChanged
153 }
154 db, err := filepath.Rel(r.Local.DataDir, in.DBPath)
155 if err != nil || !filepath.IsLocal(db) {
156 return hostbox.ErrChanged
157 }
158 db = filepath.ToSlash(db)
159 paths := c.Inventory[box].Paths
160 stamp := paths[db+".gen"]
161 if stamp == 0 || paths[db] != stamp {
162 return hostbox.ErrChanged
163 }
164 if stamp == c.Holder.Stamp {
165 return nil
166 }
167 if stamp > c.Holder.Stamp {
168 return hostbox.ErrChanged
169 }
170 if claimRecoveryKind(c.Op.Kind) {
171 sourceStamp, e := strconv.ParseUint(c.Op.Journal[hostclaim.JournalSourceStamp], 10, 64)
172 if e == nil && stamp == sourceStamp {
173 return r.Retire(ctx, admitted, in, c.Op.Journal[hostclaim.JournalSrcGen])
174 }
175 }
176 receipts, err := r.Local.Guard.CompletedWriterStops(ctx, admitted.Host, hostguard.Captured{Incarnation: c.Holder.Incarnation, Stamp: stamp}, c.Op.ID, box)
177 if err != nil {
178 return err
179 }
180 if len(receipts) == 0 {
181 return hostguard.ErrNoFenceReceipt
182 }
183 session := receipts[0].Target.Session
184 for _, a := range c.Holder.Admissions {
185 if a.Session == session && a.Box == box && a.Session < c.Holder.Session {
186 // Retire repeats all receipt, generation and guarded inventory checks;
187 // park.Local resumes the deterministic journal, even with both paths absent.
188 return r.Retire(ctx, admitted, in, a.Generation)
189 }
190 }
191 return hostbox.ErrChanged
192}
193
194func claimArtifactKind(kind string) bool {
195 return kind == hostclaim.KindLanding || claimRecoveryKind(kind)
196}
197func claimArtifactWriter(c *hostclaim.Claim, box string) (hostclaim.AdmissionRecord, bool) {
198 if c != nil && claimRecoveryKind(c.Op.Kind) {
199 return hostguard.RecoveryWriter(c, box)
200 }
201 return hostguard.LandingWriter(c, box)
202}
203func (r claimArtifactRetirement) artifactsStopped(ctx context.Context, host string, c *hostclaim.Claim, box string, stamp uint64, generation string) error {
204 sourceStamp, err := strconv.ParseUint(c.Op.Journal[hostclaim.JournalSourceStamp], 10, 64)
205 if !claimRecoveryKind(c.Op.Kind) || err != nil || stamp != sourceStamp {
206 return r.stoppedAt(ctx, host, c, box, stamp)
207 }
208 if c.Op.Intent["source"] != box || generation != c.Op.Journal[hostclaim.JournalSrcGen] {
209 return hostbox.ErrChanged
210 }
211 session, err := strconv.ParseUint(c.Op.Journal[hostclaim.JournalCutSession], 10, 64)
212 if err != nil || session >= c.Holder.Session {
213 return hostbox.ErrChanged
214 }
215 key := moves.StoppedKey(moves.ReportPrefix(c.Holder.Account, c.Holder.App, host, c.Op.ID), "source")
216 if c.Op.Journal[hostclaim.JournalEvidenceSource] != key {
217 return hostbox.ErrChanged
218 }
219 report, err := readStoppedReport(ctx, r.Local.Claims, key)
220 if err != nil {
221 return err
222 }
223 if report.Box != box || report.Role != "source" || report.Host != host || report.OpID != c.Op.ID || report.Incarnation != c.Holder.Incarnation {
224 return hostbox.ErrChanged
225 }
226 return nil
227}
228