amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package hostops
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "os"
8 "time"
9
10 "github.com/carlosframework/platform/internal/controlname"
11 "github.com/carlosframework/platform/internal/hostclaim"
12 "github.com/carlosframework/platform/internal/release"
13)
14
15type Purge struct {
16 ID string `json:"id"`
17 Phase string `json:"phase"`
18}
19
20var phases = []string{"reserved", "adopted", "claimed", "torn-down", "retired", "done"}
21
22func phaseIndex(s string) int {
23 for i, p := range phases {
24 if s == p {
25 return i
26 }
27 }
28 return -1
29}
30func PurgeKey(acct, app string) (string, error) {
31 if !controlname.ValidAccount(acct) || !controlname.ValidApp(app) {
32 return "", ErrIdentity
33 }
34 return "apps/" + acct + "/" + app + "/purge.json", nil
35}
36func ReadPurge(ctx context.Context, s release.Store, acct, app string) (Purge, string, error) {
37 key, err := PurgeKey(acct, app)
38 if err != nil {
39 return Purge{}, "", err
40 }
41 body, etag, err := s.Get(ctx, key)
42 if err != nil {
43 return Purge{}, "", err
44 }
45 var p Purge
46 if err := decode(body, &p); err != nil {
47 return Purge{}, "", err
48 }
49 if p.ID == "" || phaseIndex(p.Phase) < 0 || etag == "" {
50 return Purge{}, "", ErrIdentity
51 }
52 return p, etag, nil
53}
54func ReservePurge(ctx context.Context, s release.Store, acct, app, id string) (Purge, string, error) {
55 key, err := PurgeKey(acct, app)
56 if err != nil || id == "" {
57 return Purge{}, "", ErrIdentity
58 }
59 p := Purge{ID: id, Phase: "reserved"}
60 body, err := json.Marshal(p)
61 if err != nil {
62 return Purge{}, "", err
63 }
64 etag, err := s.Put(ctx, key, body, "")
65 if errors.Is(err, release.ErrCASConflict) {
66 existing, tag, rerr := ReadPurge(ctx, s, acct, app)
67 if rerr != nil {
68 return Purge{}, "", rerr
69 }
70 if existing.ID != id {
71 return existing, tag, ErrPending
72 }
73 return existing, tag, nil
74 }
75 return p, etag, err
76}
77
78// AdvancePurge preserves additional journal fields owned by the final durable
79// host-set schema. Preconditions for the next phase belong to the coordinator.
80func AdvancePurge(ctx context.Context, s release.Store, acct, app, id, from, to string) error {
81 if phaseIndex(from) < 0 || phaseIndex(to) != phaseIndex(from)+1 {
82 return ErrIdentity
83 }
84 key, err := PurgeKey(acct, app)
85 if err != nil {
86 return err
87 }
88 body, etag, err := s.Get(ctx, key)
89 if err != nil {
90 return err
91 }
92 var p Purge
93 if err := decode(body, &p); err != nil {
94 return err
95 }
96 if p.ID != id || p.Phase != from || etag == "" {
97 return ErrPending
98 }
99 var fields map[string]json.RawMessage
100 if err := decode(body, &fields); err != nil {
101 return err
102 }
103 fields["phase"], err = json.Marshal(to)
104 if err != nil {
105 return err
106 }
107 body, err = json.Marshal(fields)
108 if err != nil {
109 return err
110 }
111 _, err = s.Put(ctx, key, body, etag)
112 return err
113}
114
115// Only the completed coordinator may call this, after all fence-release writes.
116func RemovePurge(ctx context.Context, s release.Store, acct, app, id string) error {
117 p, etag, err := ReadPurge(ctx, s, acct, app)
118 if errors.Is(err, os.ErrNotExist) {
119 return nil
120 }
121 if err != nil {
122 return err
123 }
124 if p.ID != id || p.Phase != "done" {
125 return ErrPending
126 }
127 key, err := PurgeKey(acct, app)
128 if err != nil {
129 return err
130 }
131 return release.DeleteIfMatch(ctx, s, key, etag)
132}
133
134func FencePurge(id string, now time.Time) func(*hostclaim.Claim) error {
135 return func(c *hostclaim.Claim) error {
136 if id == "" || now.IsZero() {
137 return ErrIdentity
138 }
139 if !c.Free() || len(c.Obligations) != 0 {
140 return ErrPending
141 }
142 c.Op = hostclaim.Op{Kind: hostclaim.KindPurging, ID: id, StartedAt: now.UTC()}
143 return nil
144 }
145}
146func ReleasePurgeFence(id string) func(*hostclaim.Claim) error {
147 return func(c *hostclaim.Claim) error {
148 if id == "" || c.Holder != nil || c.Op.Kind != hostclaim.KindPurging || c.Op.ID != id || len(c.Obligations) != 0 {
149 return ErrPending
150 }
151 c.Op = hostclaim.Op{}
152 return nil
153 }
154}
155
156// BeginPurge starts one host's purge from serving. fencedCompletion is
157// hostclaim.StartGuard's (ruling 86): the completion box whose termination
158// fence the caller verified (ProductionAuthority.completionFence), or "".
159func BeginPurge(id, by string, now time.Time, regions, boxes []string, fencedCompletion string) func(*hostclaim.Claim) error {
160 return func(c *hostclaim.Claim) error {
161 if id == "" || now.IsZero() {
162 return ErrIdentity
163 }
164 if err := hostclaim.StartGuard(c, regions, boxes, fencedCompletion); err != nil {
165 return err
166 }
167 c.Op = hostclaim.Op{Kind: hostclaim.KindPurging, ID: id, By: by, StartedAt: now.UTC(), Exec: c.Op.Exec, Participants: c.Op.Participants}
168 return nil
169 }
170}
171func PurgeRecordIntent(id, key, etag string) func(*hostclaim.Claim) error {
172 return func(c *hostclaim.Claim) error {
173 if id == "" || etag == "" || c.Op.Kind != hostclaim.KindPurging || c.Op.ID != id || c.Holder == nil || c.Holder.Key != key {
174 return ErrIdentity
175 }
176 if c.Op.Intent == nil {
177 c.Op.Intent = map[string]string{}
178 }
179 if old := c.Op.Intent["etag"]; old != "" && old != etag {
180 return ErrPending
181 }
182 c.Op.Intent["key"], c.Op.Intent["etag"] = key, etag
183 return nil
184 }
185}
186
187// No completion transition is supplied until placement's executor/evidence
188// contract is resolved. This start transition alone grants no new execution.
189func BeginPlacement(id, by, target string, now time.Time, regions, boxes []string) func(*hostclaim.Claim) error {
190 return func(c *hostclaim.Claim) error {
191 if id == "" || now.IsZero() {
192 return ErrIdentity
193 }
194 if err := hostclaim.StartGuard(c, regions, boxes, ""); err != nil {
195 return err
196 }
197 c.Op = hostclaim.Op{Kind: hostclaim.KindPlacing, ID: id, By: by, StartedAt: now.UTC(), Exec: c.Op.Exec, Participants: c.Op.Participants, Intent: map[string]string{"target": target}}
198 return nil
199 }
200}
201
202func CompleteMetadataPlacement(id, appID string) func(*hostclaim.Claim) error {
203 return func(c *hostclaim.Claim) error {
204 if id == "" || appID == "" || c.Holder == nil || c.Op.Kind != hostclaim.KindPlacing || c.Op.ID != id || c.Op.Intent["app_id"] != appID {
205 return ErrPending
206 }
207 c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: c.Op.Exec, Participants: c.Op.Participants}
208 return nil
209 }
210}
211