| 1 | package hostexec |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "crypto/rand" |
| 7 | "crypto/sha256" |
| 8 | "encoding/hex" |
| 9 | "encoding/json" |
| 10 | "errors" |
| 11 | "os" |
| 12 | "strings" |
| 13 | "sync" |
| 14 | "time" |
| 15 | |
| 16 | "github.com/carlosframework/platform/internal/release" |
| 17 | "github.com/carlosframework/platform/internal/rootparticipation" |
| 18 | "github.com/carlosframework/platform/internal/rootstate" |
| 19 | ) |
| 20 | |
| 21 | type ExecutionRequest = rootparticipation.ExecutionRequest |
| 22 | type ExecutionAuthority = rootparticipation.ExecutionAuthority |
| 23 | type ExecutionResponse = rootparticipation.ExecutionResponse |
| 24 | type ExecutionReplay = rootparticipation.ExecutionReplay |
| 25 | type ExecutionRecord = rootparticipation.ExecutionRecord |
| 26 | |
| 27 | type RootBrokerConfig struct { |
| 28 | Recovered func(context.Context) error |
| 29 | Transition func(context.Context, ExecutionRecord, ExecutionRequest, ExecutionAuthority) error |
| 30 | State rootstate.Store |
| 31 | Remote release.Store |
| 32 | Clock Clock |
| 33 | BootID, Identity string |
| 34 | Authority func(context.Context, ExecutionRequest, ControllerProcess) (ExecutionAuthority, error) |
| 35 | Alive func(context.Context, ControllerProcess) error |
| 36 | Cleanup func(context.Context, ExecutionRecord) error |
| 37 | Settle func(context.Context, ExecutionRecord) error |
| 38 | Exclude func(context.Context, func() error) error |
| 39 | } |
| 40 | |
| 41 | type RootBroker struct { |
| 42 | cfg RootBrokerConfig |
| 43 | mu sync.Mutex |
| 44 | workers sync.WaitGroup |
| 45 | stopping bool |
| 46 | disposed bool |
| 47 | joined chan struct{} |
| 48 | closeGate chan struct{} |
| 49 | lifetimeCtx context.Context |
| 50 | cancel context.CancelFunc |
| 51 | closeErr error |
| 52 | denied map[string]bool |
| 53 | closed bool |
| 54 | calls map[string]chan struct{} |
| 55 | lifetime map[string]func() |
| 56 | live map[string]func() |
| 57 | retryAfter map[string]int64 |
| 58 | retryDelay map[string]time.Duration |
| 59 | } |
| 60 | |
| 61 | func NewRootBroker(cfg RootBrokerConfig) (*RootBroker, error) { |
| 62 | if cfg.State == nil || cfg.Remote == nil || cfg.Clock == nil || !validID(cfg.BootID) || !validID(cfg.Identity) || cfg.Authority == nil || cfg.Alive == nil || cfg.Cleanup == nil || cfg.Settle == nil || cfg.Exclude == nil { |
| 63 | return nil, ErrState |
| 64 | } |
| 65 | lifetimeCtx, cancel := context.WithCancel(context.Background()) |
| 66 | return &RootBroker{cfg: cfg, lifetimeCtx: lifetimeCtx, cancel: cancel, closeGate: make(chan struct{}, 1), denied: make(map[string]bool), calls: make(map[string]chan struct{}), lifetime: make(map[string]func()), live: make(map[string]func()), retryAfter: make(map[string]int64), retryDelay: make(map[string]time.Duration)}, nil |
| 67 | } |
| 68 | |
| 69 | func (b *RootBroker) enter(ctx context.Context) (context.Context, func(), error) { |
| 70 | b.mu.Lock() |
| 71 | if b.stopping { |
| 72 | b.mu.Unlock() |
| 73 | return nil, nil, ErrLost |
| 74 | } |
| 75 | b.workers.Add(1) |
| 76 | b.mu.Unlock() |
| 77 | work, cancel := context.WithCancel(ctx) |
| 78 | stop := context.AfterFunc(b.lifetimeCtx, cancel) |
| 79 | return work, func() { stop(); cancel(); b.workers.Done() }, nil |
| 80 | } |
| 81 | |
| 82 | func executionDigest(q ExecutionRequest, p ControllerProcess) string { |
| 83 | b, _ := json.Marshal(struct { |
| 84 | Request ExecutionRequest |
| 85 | Peer ControllerProcess |
| 86 | }{q, p}) |
| 87 | s := sha256.Sum256(b) |
| 88 | return hex.EncodeToString(s[:]) |
| 89 | } |
| 90 | |
| 91 | func (b *RootBroker) read(host string) (ExecutionRecord, error) { |
| 92 | b.mu.Lock() |
| 93 | defer b.mu.Unlock() |
| 94 | raw, err := b.cfg.State.Read(host + ".json") |
| 95 | if err != nil { |
| 96 | return ExecutionRecord{}, err |
| 97 | } |
| 98 | var r ExecutionRecord |
| 99 | if decode(raw, &r) != nil || r.Schema != 1 || r.Request.Host != host || !r.Request.Valid() || !validID(r.Token) || r.Authority.Migration == 0 || r.Replay == nil { |
| 100 | return ExecutionRecord{}, ErrState |
| 101 | } |
| 102 | switch r.Status { |
| 103 | case "pending", "active", "lost", "settled": |
| 104 | default: |
| 105 | return ExecutionRecord{}, ErrState |
| 106 | } |
| 107 | return r, nil |
| 108 | } |
| 109 | |
| 110 | func (b *RootBroker) write(r ExecutionRecord) error { |
| 111 | b.mu.Lock() |
| 112 | defer b.mu.Unlock() |
| 113 | if r.Status == "active" && (b.closed || b.denied[r.Request.Host]) { |
| 114 | return ErrLost |
| 115 | } |
| 116 | if release := b.live[r.Request.Host]; release != nil { |
| 117 | release() |
| 118 | delete(b.live, r.Request.Host) |
| 119 | } |
| 120 | raw, err := json.Marshal(r) |
| 121 | if err != nil { |
| 122 | return err |
| 123 | } |
| 124 | if err := b.cfg.State.Write(r.Request.Host+".json", raw); err != nil { |
| 125 | return err |
| 126 | } |
| 127 | if r.Status == "active" { |
| 128 | release, err := rootstate.HoldLive(b.cfg.State, r.Request.Host+".live") |
| 129 | if err != nil { |
| 130 | return err |
| 131 | } |
| 132 | b.live[r.Request.Host] = release |
| 133 | } |
| 134 | return nil |
| 135 | } |
| 136 | |
| 137 | func executionResponse(r ExecutionRecord) ExecutionResponse { |
| 138 | q := r.Request |
| 139 | return ExecutionResponse{Schema: 1, Host: q.Host, Incarnation: q.Incarnation, Stamp: q.Stamp, Session: q.Session, Op: q.Op, Token: r.Token, BootID: r.Peer.KernelBootID, Status: r.Status, SafeUntil: r.SafeUntil, LeaseETag: r.LeaseETag, Expires: r.Expires} |
| 140 | } |
| 141 | |
| 142 | func (b *RootBroker) active(ctx context.Context, r ExecutionRecord) error { |
| 143 | b.mu.Lock() |
| 144 | denied := b.closed || b.denied[r.Request.Host] |
| 145 | b.mu.Unlock() |
| 146 | if denied || r.Status != "active" || r.Broker != b.cfg.Identity || r.Peer.KernelBootID != b.cfg.BootID { |
| 147 | return ErrLost |
| 148 | } |
| 149 | _, now, err := b.cfg.Clock.Now() |
| 150 | if err != nil { |
| 151 | return err |
| 152 | } |
| 153 | if now >= r.SafeUntil { |
| 154 | return ErrLost |
| 155 | } |
| 156 | return b.cfg.Alive(ctx, r.Peer) |
| 157 | } |
| 158 | |
| 159 | func sameExecution(q ExecutionRequest, r ExecutionRecord) bool { |
| 160 | p := r.Request |
| 161 | return q.Host == p.Host && q.Incarnation == p.Incarnation && q.Stamp == p.Stamp && q.Session == p.Session && q.Op == p.Op |
| 162 | } |
| 163 | |
| 164 | func (b *RootBroker) deny(ctx context.Context, r ExecutionRecord, cause error) error { |
| 165 | b.mu.Lock() |
| 166 | raw, readErr := b.cfg.State.Read(r.Request.Host + ".json") |
| 167 | var current ExecutionRecord |
| 168 | if readErr == nil && decode(raw, ¤t) == nil && ((current.Token != r.Token && current.Token != r.TransitionFrom) || current.Status == "settled") { |
| 169 | b.mu.Unlock() |
| 170 | return cause |
| 171 | } |
| 172 | if readErr == nil && current.Token == r.Token && current.RequestStart > r.RequestStart { |
| 173 | r = current |
| 174 | } |
| 175 | b.denied[r.Request.Host] = true |
| 176 | if release := b.live[r.Request.Host]; release != nil { |
| 177 | release() |
| 178 | delete(b.live, r.Request.Host) |
| 179 | } |
| 180 | r.Status = "lost" |
| 181 | raw, persist := json.Marshal(r) |
| 182 | if persist == nil { |
| 183 | persist = b.cfg.State.Write(r.Request.Host+".json", raw) |
| 184 | } |
| 185 | if persist != nil { |
| 186 | b.closed = true |
| 187 | for host, release := range b.live { |
| 188 | release() |
| 189 | delete(b.live, host) |
| 190 | } |
| 191 | } |
| 192 | b.mu.Unlock() |
| 193 | cleanup, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) |
| 194 | defer cancel() |
| 195 | clean := b.cfg.Cleanup(cleanup, r) |
| 196 | return errors.Join(cause, persist, clean) |
| 197 | } |
| 198 | |
| 199 | func (b *RootBroker) Execute(ctx context.Context, peer ControllerProcess, q ExecutionRequest) (ExecutionResponse, error) { |
| 200 | if !q.Valid() || peer.PID <= 0 || !validID(peer.Start) || peer.KernelBootID != b.cfg.BootID { |
| 201 | return ExecutionResponse{}, ErrState |
| 202 | } |
| 203 | ctx, leave, enterErr := b.enter(ctx) |
| 204 | if enterErr != nil { |
| 205 | return ExecutionResponse{}, enterErr |
| 206 | } |
| 207 | defer leave() |
| 208 | if q.Operation != "executionSettle" { |
| 209 | if err := b.cfg.Alive(ctx, peer); err != nil { |
| 210 | return ExecutionResponse{}, err |
| 211 | } |
| 212 | } |
| 213 | b.mu.Lock() |
| 214 | gate := b.calls[q.Host] |
| 215 | if gate == nil { |
| 216 | if len(b.calls) >= 4096 { |
| 217 | b.mu.Unlock() |
| 218 | return ExecutionResponse{}, ErrState |
| 219 | } |
| 220 | gate = make(chan struct{}, 1) |
| 221 | b.calls[q.Host] = gate |
| 222 | } |
| 223 | b.mu.Unlock() |
| 224 | select { |
| 225 | case gate <- struct{}{}: |
| 226 | case <-ctx.Done(): |
| 227 | return ExecutionResponse{}, ctx.Err() |
| 228 | } |
| 229 | defer func() { <-gate }() |
| 230 | b.mu.Lock() |
| 231 | held := b.lifetime[q.Host] != nil |
| 232 | b.mu.Unlock() |
| 233 | if !held { |
| 234 | unlock, lockErr := b.cfg.State.Lock(ctx, q.Host+".lock") |
| 235 | if lockErr != nil { |
| 236 | return ExecutionResponse{}, lockErr |
| 237 | } |
| 238 | b.mu.Lock() |
| 239 | b.lifetime[q.Host] = unlock |
| 240 | b.mu.Unlock() |
| 241 | } |
| 242 | if err := ctx.Err(); err != nil { |
| 243 | return ExecutionResponse{}, err |
| 244 | } |
| 245 | // Settlement verifies retained obligations even after current admission ends. |
| 246 | var authority ExecutionAuthority |
| 247 | var err error |
| 248 | if q.Operation != "executionSettle" { |
| 249 | cutoff := int64(0) |
| 250 | if prior, readErr := b.read(q.Host); readErr == nil && prior.Status == "active" { |
| 251 | cutoff = prior.SafeUntil |
| 252 | } |
| 253 | err = b.retry(ctx, cutoff, func(call context.Context) error { var e error; authority, e = b.cfg.Authority(call, q, peer); return e }) |
| 254 | if err == nil && (authority.Migration == 0 || (authority.Installation == "" && !authority.ParticipationOnly) || !validID(authority.Box) || !validID(authority.Enrollment) || authority.Account == "" || authority.App == "" || authority.ClaimRevision == "" || authority.RecordRevision == "" || authority.DirectoryRevision == "") { |
| 255 | err = ErrState |
| 256 | } |
| 257 | if err != nil { |
| 258 | if prior, readErr := b.read(q.Host); readErr == nil && prior.Peer == peer && (q.Token == prior.Token || q.ID == prior.Request.ID) && prior.Status != "settled" { |
| 259 | return ExecutionResponse{}, b.deny(ctx, prior, err) |
| 260 | } |
| 261 | return ExecutionResponse{}, err |
| 262 | } |
| 263 | if q.Operation == "executionView" && q.Token == "" { |
| 264 | return b.servingView(ctx, peer, q, authority) |
| 265 | } |
| 266 | if authority.Kind == "serving" { |
| 267 | return ExecutionResponse{}, ErrInactive |
| 268 | } |
| 269 | } |
| 270 | r, err := b.read(q.Host) |
| 271 | missing := errors.Is(err, os.ErrNotExist) |
| 272 | if err != nil && !missing { |
| 273 | return ExecutionResponse{}, err |
| 274 | } |
| 275 | digest := executionDigest(q, peer) |
| 276 | if !missing { |
| 277 | if replay, ok := r.Replay[q.ID]; ok { |
| 278 | if replay.Digest != digest { |
| 279 | return ExecutionResponse{}, ErrState |
| 280 | } |
| 281 | if q.Operation == "executionSettle" && replay.Response.Status == "settled" { |
| 282 | return replay.Response, nil |
| 283 | } |
| 284 | if !sameExecution(q, r) || replay.Response.Token != r.Token { |
| 285 | return ExecutionResponse{}, ErrLost |
| 286 | } |
| 287 | if err = b.active(ctx, r); err != nil { |
| 288 | return ExecutionResponse{}, b.deny(ctx, r, err) |
| 289 | } |
| 290 | if r.Peer != peer || r.Authority != authority { |
| 291 | return ExecutionResponse{}, b.deny(ctx, r, ErrLost) |
| 292 | } |
| 293 | return replay.Response, nil |
| 294 | } |
| 295 | } |
| 296 | if q.Operation == "executionBegin" { |
| 297 | b.mu.Lock() |
| 298 | closed := b.closed |
| 299 | b.mu.Unlock() |
| 300 | if closed || (!missing && (r.Status != "settled" || sameExecution(q, r))) { |
| 301 | return ExecutionResponse{}, ErrLost |
| 302 | } |
| 303 | var token [32]byte |
| 304 | if _, err = rand.Read(token[:]); err != nil { |
| 305 | return ExecutionResponse{}, err |
| 306 | } |
| 307 | history := make(map[string]ExecutionReplay) |
| 308 | for id, replay := range r.Replay { |
| 309 | if replay.Response.Status == "settled" && replay.Response.Token == r.Token { |
| 310 | history[id] = replay |
| 311 | } |
| 312 | } |
| 313 | b.mu.Lock() |
| 314 | delete(b.denied, q.Host) |
| 315 | delete(b.retryAfter, q.Host) |
| 316 | delete(b.retryDelay, q.Host) |
| 317 | b.mu.Unlock() |
| 318 | r = ExecutionRecord{Schema: 1, Request: q, Peer: peer, Authority: authority, Broker: b.cfg.Identity, Token: hex.EncodeToString(token[:]), Status: "pending", Replay: history} |
| 319 | wall, start, e := b.cfg.Clock.Now() |
| 320 | if e != nil { |
| 321 | return ExecutionResponse{}, e |
| 322 | } |
| 323 | r.PriorSafeUntil = r.SafeUntil |
| 324 | r.RequestStart = start |
| 325 | r.SafeUntil = start + int64(TTL-Margin) |
| 326 | r.Expires = wall.Add(TTL) |
| 327 | r.Replay[q.ID] = ExecutionReplay{Digest: digest} |
| 328 | if err = b.cfg.Exclude(ctx, func() error { return b.write(r) }); err != nil { |
| 329 | return ExecutionResponse{}, err |
| 330 | } |
| 331 | return b.acquire(ctx, peer, q, r, digest, true) |
| 332 | } |
| 333 | if missing || r.Peer != peer || q.Token != r.Token { |
| 334 | return ExecutionResponse{}, ErrLost |
| 335 | } |
| 336 | if !sameExecution(q, r) { |
| 337 | if q.Operation != "executionRenew" || b.cfg.Transition == nil || q.Host != r.Request.Host || q.Op != r.Request.Op || q.Incarnation != r.Request.Incarnation { |
| 338 | return ExecutionResponse{}, ErrLost |
| 339 | } |
| 340 | if err = b.active(ctx, r); err != nil { |
| 341 | return ExecutionResponse{}, b.deny(ctx, r, err) |
| 342 | } |
| 343 | if err = b.cfg.Transition(ctx, r, q, authority); err != nil { |
| 344 | return ExecutionResponse{}, b.deny(ctx, r, err) |
| 345 | } |
| 346 | var token [32]byte |
| 347 | if _, err = rand.Read(token[:]); err != nil { |
| 348 | return ExecutionResponse{}, err |
| 349 | } |
| 350 | r.Request = q |
| 351 | r.TransitionFrom = r.Token |
| 352 | r.Token = hex.EncodeToString(token[:]) |
| 353 | r.Authority = authority |
| 354 | return b.acquire(ctx, peer, q, r, digest, false) |
| 355 | } |
| 356 | if q.Operation == "executionSettle" { |
| 357 | return b.settle(ctx, q, r, digest) |
| 358 | } |
| 359 | if err = b.active(ctx, r); err != nil { |
| 360 | return ExecutionResponse{}, b.deny(ctx, r, err) |
| 361 | } |
| 362 | if authority != r.Authority { |
| 363 | if !executionInstallationBinding(r.Authority, authority) { |
| 364 | if (q.Operation != "executionRenew" && q.Operation != "executionView") || b.cfg.Transition == nil || b.cfg.Transition(ctx, r, q, authority) != nil { |
| 365 | return ExecutionResponse{}, b.deny(ctx, r, ErrLost) |
| 366 | } |
| 367 | r.Authority = authority |
| 368 | if q.Operation == "executionView" { |
| 369 | if err = b.write(r); err != nil { |
| 370 | return ExecutionResponse{}, b.deny(ctx, r, err) |
| 371 | } |
| 372 | return executionResponse(r), nil |
| 373 | } |
| 374 | return b.acquire(ctx, peer, q, r, digest, false) |
| 375 | } |
| 376 | r.Authority = authority |
| 377 | if err = b.write(r); err != nil { |
| 378 | return ExecutionResponse{}, b.deny(ctx, r, err) |
| 379 | } |
| 380 | } |
| 381 | if q.Operation == "executionView" { |
| 382 | return executionResponse(r), nil |
| 383 | } |
| 384 | return b.acquire(ctx, peer, q, r, digest, false) |
| 385 | } |
| 386 | |
| 387 | func (b *RootBroker) acquire(ctx context.Context, peer ControllerProcess, q ExecutionRequest, r ExecutionRecord, digest string, begin bool) (ExecutionResponse, error) { |
| 388 | key, err := Key(q.Host, r.Authority.Box) |
| 389 | if err != nil { |
| 390 | return ExecutionResponse{}, err |
| 391 | } |
| 392 | etag := r.LeaseETag |
| 393 | if begin { |
| 394 | raw, existing, e := b.cfg.Remote.Get(ctx, key) |
| 395 | if e == nil { |
| 396 | var remote Record |
| 397 | wall, _, clockErr := b.cfg.Clock.Now() |
| 398 | if decode(raw, &remote) != nil || remote.Expires.IsZero() || !wall.After(remote.Expires) || existing == "" || clockErr != nil { |
| 399 | return ExecutionResponse{}, b.deny(ctx, r, ErrLost) |
| 400 | } |
| 401 | etag = existing |
| 402 | } else if !errors.Is(e, os.ErrNotExist) { |
| 403 | return ExecutionResponse{}, b.deny(ctx, r, e) |
| 404 | } |
| 405 | } else { |
| 406 | if r.Token == q.Token { |
| 407 | var token [32]byte |
| 408 | if _, err := rand.Read(token[:]); err != nil { |
| 409 | return ExecutionResponse{}, err |
| 410 | } |
| 411 | r.TransitionFrom, r.Token = r.Token, hex.EncodeToString(token[:]) |
| 412 | } |
| 413 | priorID := "" |
| 414 | var prior ExecutionReplay |
| 415 | for id, replay := range r.Replay { |
| 416 | if replay.Response.Status == "settled" && (priorID == "" || replay.Response.Expires.After(prior.Response.Expires) || replay.Response.Expires.Equal(prior.Response.Expires) && id > priorID) { |
| 417 | priorID, prior = id, replay |
| 418 | } |
| 419 | } |
| 420 | r.Replay = make(map[string]ExecutionReplay) |
| 421 | if priorID != "" { |
| 422 | r.Replay[priorID] = prior |
| 423 | } |
| 424 | wall, start, e := b.cfg.Clock.Now() |
| 425 | if e != nil { |
| 426 | return ExecutionResponse{}, b.deny(ctx, r, e) |
| 427 | } |
| 428 | r.PriorSafeUntil = r.SafeUntil |
| 429 | r.RequestStart = start |
| 430 | r.SafeUntil = start + int64(TTL-Margin) |
| 431 | r.Expires = wall.Add(TTL) |
| 432 | r.Status = "pending" |
| 433 | r.Replay[q.ID] = ExecutionReplay{Digest: digest} |
| 434 | if e = b.cfg.Exclude(ctx, func() error { return b.write(r) }); e != nil { |
| 435 | return ExecutionResponse{}, b.deny(ctx, r, e) |
| 436 | } |
| 437 | } |
| 438 | raw, _ := json.Marshal(Record{Op: q.Op, Expires: r.Expires}) |
| 439 | var next string |
| 440 | cutoff := r.SafeUntil |
| 441 | if r.PriorSafeUntil > 0 && r.PriorSafeUntil < cutoff { |
| 442 | cutoff = r.PriorSafeUntil |
| 443 | } |
| 444 | uncertain := false |
| 445 | err = b.retry(ctx, cutoff, func(call context.Context) error { |
| 446 | if uncertain { |
| 447 | found, current, e := b.cfg.Remote.Get(call, key) |
| 448 | if e == nil && bytes.Equal(found, raw) && current != "" { |
| 449 | next = current |
| 450 | return nil |
| 451 | } |
| 452 | if e != nil && !errors.Is(e, os.ErrNotExist) { |
| 453 | return e |
| 454 | } |
| 455 | if e == nil && current != etag { |
| 456 | return ErrLost |
| 457 | } |
| 458 | } |
| 459 | var e error |
| 460 | next, e = b.cfg.Remote.Put(call, key, raw, etag) |
| 461 | if e != nil && !errors.Is(e, release.ErrCASConflict) { |
| 462 | uncertain = true |
| 463 | } |
| 464 | return e |
| 465 | }) |
| 466 | if err != nil || next == "" { |
| 467 | return ExecutionResponse{}, b.deny(ctx, r, errors.Join(err, ErrLost)) |
| 468 | } |
| 469 | r.LeaseETag = next |
| 470 | var authority ExecutionAuthority |
| 471 | err = b.retry(ctx, cutoff, func(call context.Context) error { var e error; authority, e = b.cfg.Authority(call, q, peer); return e }) |
| 472 | if err != nil || authority != r.Authority { |
| 473 | return ExecutionResponse{}, b.deny(ctx, r, errors.Join(err, ErrLost)) |
| 474 | } |
| 475 | _, now, err := b.cfg.Clock.Now() |
| 476 | if err != nil || now >= r.SafeUntil || (!begin && r.PriorSafeUntil > 0 && now >= r.PriorSafeUntil) || ctx.Err() != nil { |
| 477 | return ExecutionResponse{}, b.deny(ctx, r, errors.Join(err, ErrLost)) |
| 478 | } |
| 479 | if err = b.cfg.Alive(ctx, peer); err != nil { |
| 480 | return ExecutionResponse{}, b.deny(ctx, r, err) |
| 481 | } |
| 482 | r.PriorSafeUntil = 0 |
| 483 | r.Status = "active" |
| 484 | out := executionResponse(r) |
| 485 | r.Replay[q.ID] = ExecutionReplay{Digest: digest, Response: out} |
| 486 | if err = b.write(r); err != nil { |
| 487 | return ExecutionResponse{}, b.deny(ctx, r, err) |
| 488 | } |
| 489 | if err = b.active(ctx, r); err != nil { |
| 490 | return ExecutionResponse{}, b.deny(ctx, r, err) |
| 491 | } |
| 492 | return out, nil |
| 493 | } |
| 494 | |
| 495 | func (b *RootBroker) settle(ctx context.Context, q ExecutionRequest, r ExecutionRecord, digest string) (ExecutionResponse, error) { |
| 496 | if r.Status == "settled" { |
| 497 | return executionResponse(r), nil |
| 498 | } |
| 499 | if err := b.deny(ctx, r, nil); err != nil { |
| 500 | return ExecutionResponse{}, err |
| 501 | } |
| 502 | r.Status = "lost" |
| 503 | key, err := Key(q.Host, r.Authority.Box) |
| 504 | if err != nil { |
| 505 | return ExecutionResponse{}, err |
| 506 | } |
| 507 | deletionUncertain := r.LeaseETag == "" |
| 508 | if !deletionUncertain { |
| 509 | err = release.DeleteIfMatch(ctx, b.cfg.Remote, key, r.LeaseETag) |
| 510 | if errors.Is(err, release.ErrCASConflict) { |
| 511 | deletionUncertain = true |
| 512 | } else if err != nil && !errors.Is(err, os.ErrNotExist) { |
| 513 | return ExecutionResponse{}, err |
| 514 | } |
| 515 | } |
| 516 | if deletionUncertain { |
| 517 | raw, currentETag, readErr := b.cfg.Remote.Get(ctx, key) |
| 518 | wall, _, clockErr := b.cfg.Clock.Now() |
| 519 | if clockErr != nil { |
| 520 | return ExecutionResponse{}, clockErr |
| 521 | } |
| 522 | if !wall.After(r.Expires) { |
| 523 | return ExecutionResponse{}, ErrLost |
| 524 | } |
| 525 | if readErr == nil { |
| 526 | var remote Record |
| 527 | if decode(raw, &remote) != nil || remote.Expires.IsZero() || !wall.After(remote.Expires) || currentETag == "" { |
| 528 | return ExecutionResponse{}, ErrLost |
| 529 | } |
| 530 | if err := release.DeleteIfMatch(ctx, b.cfg.Remote, key, currentETag); err != nil { |
| 531 | return ExecutionResponse{}, err |
| 532 | } |
| 533 | } else if !errors.Is(readErr, os.ErrNotExist) { |
| 534 | return ExecutionResponse{}, readErr |
| 535 | } |
| 536 | } |
| 537 | if err = b.cfg.Settle(ctx, r); err != nil { |
| 538 | return ExecutionResponse{}, err |
| 539 | } |
| 540 | r.Status = "settled" |
| 541 | out := executionResponse(r) |
| 542 | r.Replay = make(map[string]ExecutionReplay) |
| 543 | r.Replay[q.ID] = ExecutionReplay{Digest: digest, Response: out} |
| 544 | if err = b.write(r); err != nil { |
| 545 | return ExecutionResponse{}, err |
| 546 | } |
| 547 | return out, nil |
| 548 | } |
| 549 | |
| 550 | func (b *RootBroker) Monitor(ctx context.Context) error { |
| 551 | ctx, leave, err := b.enter(ctx) |
| 552 | if err != nil { |
| 553 | return err |
| 554 | } |
| 555 | defer leave() |
| 556 | names, err := b.cfg.State.Names() |
| 557 | if err != nil { |
| 558 | return err |
| 559 | } |
| 560 | var errs []error |
| 561 | for _, name := range names { |
| 562 | if name == "census.json" || name == "update.json" || !strings.HasSuffix(name, ".json") { |
| 563 | continue |
| 564 | } |
| 565 | host := strings.TrimSuffix(name, ".json") |
| 566 | b.mu.Lock() |
| 567 | owned := b.lifetime[host] != nil |
| 568 | b.mu.Unlock() |
| 569 | if !owned { |
| 570 | continue |
| 571 | } |
| 572 | r, err := b.read(host) |
| 573 | if err != nil { |
| 574 | errs = append(errs, err) |
| 575 | continue |
| 576 | } |
| 577 | if r.Status == "settled" { |
| 578 | continue |
| 579 | } |
| 580 | if r.Status == "lost" { |
| 581 | _, now, clockErr := b.cfg.Clock.Now() |
| 582 | if clockErr != nil { |
| 583 | errs = append(errs, clockErr) |
| 584 | continue |
| 585 | } |
| 586 | b.mu.Lock() |
| 587 | next := b.retryAfter[host] |
| 588 | b.mu.Unlock() |
| 589 | if now < next { |
| 590 | continue |
| 591 | } |
| 592 | err := b.recoverSettlement(ctx, host) |
| 593 | b.mu.Lock() |
| 594 | if err != nil { |
| 595 | delay := b.retryDelay[host] * 2 |
| 596 | if delay < time.Second { |
| 597 | delay = time.Second |
| 598 | } |
| 599 | if delay > 30*time.Second { |
| 600 | delay = 30 * time.Second |
| 601 | } |
| 602 | b.retryDelay[host], b.retryAfter[host] = delay, now+int64(delay) |
| 603 | } else { |
| 604 | delete(b.retryDelay, host) |
| 605 | delete(b.retryAfter, host) |
| 606 | } |
| 607 | b.mu.Unlock() |
| 608 | errs = append(errs, err) |
| 609 | continue |
| 610 | } |
| 611 | _, now, clockErr := b.cfg.Clock.Now() |
| 612 | bounded, cancel := context.WithTimeout(ctx, time.Second) |
| 613 | aliveErr := b.cfg.Alive(bounded, r.Peer) |
| 614 | cancel() |
| 615 | cutoff := r.SafeUntil |
| 616 | if r.Status == "pending" && r.PriorSafeUntil > 0 && r.PriorSafeUntil < cutoff { |
| 617 | cutoff = r.PriorSafeUntil |
| 618 | } |
| 619 | if clockErr != nil || aliveErr != nil || now >= cutoff || r.Broker != b.cfg.Identity || r.Peer.KernelBootID != b.cfg.BootID { |
| 620 | errs = append(errs, b.deny(ctx, r, ErrLost)) |
| 621 | } |
| 622 | } |
| 623 | if b.cfg.Recovered != nil { |
| 624 | errs = append(errs, b.cfg.Recovered(ctx)) |
| 625 | } |
| 626 | return errors.Join(errs...) |
| 627 | } |
| 628 | |
| 629 | func executionInstallationBinding(old, next ExecutionAuthority) bool { |
| 630 | if !old.ParticipationOnly || (old.Installation != "" && (old.Installation != next.Installation || next.ParticipationOnly)) || next.Installation == "" { |
| 631 | return false |
| 632 | } |
| 633 | old.Installation = next.Installation |
| 634 | old.ParticipationOnly = next.ParticipationOnly |
| 635 | return old == next |
| 636 | } |
| 637 | |
| 638 | func (b *RootBroker) Close(ctx context.Context) error { |
| 639 | b.mu.Lock() |
| 640 | b.closed = true |
| 641 | if !b.stopping { |
| 642 | b.stopping = true |
| 643 | b.cancel() |
| 644 | for host, unlock := range b.live { |
| 645 | unlock() |
| 646 | delete(b.live, host) |
| 647 | } |
| 648 | b.joined = make(chan struct{}) |
| 649 | go func() { b.workers.Wait(); close(b.joined) }() |
| 650 | } |
| 651 | joined := b.joined |
| 652 | b.mu.Unlock() |
| 653 | select { |
| 654 | case <-joined: |
| 655 | case <-ctx.Done(): |
| 656 | return ctx.Err() |
| 657 | } |
| 658 | select { |
| 659 | case b.closeGate <- struct{}{}: |
| 660 | case <-ctx.Done(): |
| 661 | return ctx.Err() |
| 662 | } |
| 663 | defer func() { <-b.closeGate }() |
| 664 | if b.disposed { |
| 665 | return b.closeErr |
| 666 | } |
| 667 | names, err := b.cfg.State.Names() |
| 668 | errs := []error{err} |
| 669 | for _, name := range names { |
| 670 | if name == "census.json" || name == "update.json" || !strings.HasSuffix(name, ".json") { |
| 671 | continue |
| 672 | } |
| 673 | r, e := b.read(strings.TrimSuffix(name, ".json")) |
| 674 | if e != nil { |
| 675 | errs = append(errs, e) |
| 676 | continue |
| 677 | } |
| 678 | if r.Broker == b.cfg.Identity && r.Status != "settled" { |
| 679 | errs = append(errs, b.deny(ctx, r, nil)) |
| 680 | } |
| 681 | } |
| 682 | b.mu.Lock() |
| 683 | for host, unlock := range b.lifetime { |
| 684 | unlock() |
| 685 | delete(b.lifetime, host) |
| 686 | } |
| 687 | b.mu.Unlock() |
| 688 | b.disposed = true |
| 689 | b.closeErr = errors.Join(errs...) |
| 690 | return b.closeErr |
| 691 | } |
| 692 | |
| 693 | func (b *RootBroker) Recover(ctx context.Context) error { |
| 694 | ctx, leave, err := b.enter(ctx) |
| 695 | if err != nil { |
| 696 | return err |
| 697 | } |
| 698 | defer leave() |
| 699 | names, err := b.cfg.State.Names() |
| 700 | if err != nil { |
| 701 | return err |
| 702 | } |
| 703 | var errs []error |
| 704 | for _, name := range names { |
| 705 | if name == "census.json" || name == "update.json" || !strings.HasSuffix(name, ".json") { |
| 706 | continue |
| 707 | } |
| 708 | host := strings.TrimSuffix(name, ".json") |
| 709 | b.mu.Lock() |
| 710 | owned := b.lifetime[host] != nil |
| 711 | b.mu.Unlock() |
| 712 | if !owned { |
| 713 | unlock, e := b.cfg.State.Lock(ctx, host+".lock") |
| 714 | if e != nil { |
| 715 | return e |
| 716 | } |
| 717 | b.mu.Lock() |
| 718 | b.lifetime[host] = unlock |
| 719 | b.mu.Unlock() |
| 720 | } |
| 721 | r, e := b.read(host) |
| 722 | if e != nil { |
| 723 | errs = append(errs, e) |
| 724 | continue |
| 725 | } |
| 726 | if r.Status != "settled" { |
| 727 | errs = append(errs, b.recoverSettlement(ctx, host)) |
| 728 | } |
| 729 | } |
| 730 | if b.cfg.Recovered != nil { |
| 731 | errs = append(errs, b.cfg.Recovered(ctx)) |
| 732 | } |
| 733 | return errors.Join(errs...) |
| 734 | } |
| 735 | |
| 736 | func (b *RootBroker) recoverSettlement(ctx context.Context, host string) error { |
| 737 | b.mu.Lock() |
| 738 | gate := b.calls[host] |
| 739 | if gate == nil { |
| 740 | gate = make(chan struct{}, 1) |
| 741 | b.calls[host] = gate |
| 742 | } |
| 743 | b.mu.Unlock() |
| 744 | select { |
| 745 | case gate <- struct{}{}: |
| 746 | default: |
| 747 | return nil |
| 748 | } |
| 749 | defer func() { <-gate }() |
| 750 | r, err := b.read(host) |
| 751 | if err != nil || r.Status == "settled" { |
| 752 | return err |
| 753 | } |
| 754 | q := r.Request |
| 755 | q.Operation, q.Token, q.ID = "executionSettle", r.Token, "recovery-"+r.Token |
| 756 | _, err = b.settle(ctx, q, r, executionDigest(q, r.Peer)) |
| 757 | return err |
| 758 | } |
| 759 | |
| 760 | func (b *RootBroker) Run(ctx context.Context) { |
| 761 | ctx, leave, err := b.enter(ctx) |
| 762 | if err != nil { |
| 763 | return |
| 764 | } |
| 765 | defer leave() |
| 766 | for { |
| 767 | select { |
| 768 | case <-ctx.Done(): |
| 769 | return |
| 770 | case <-b.lifetimeCtx.Done(): |
| 771 | return |
| 772 | case <-b.cfg.Clock.After(100 * time.Millisecond): |
| 773 | } |
| 774 | _ = b.Monitor(ctx) |
| 775 | } |
| 776 | } |
| 777 | |
| 778 | func (b *RootBroker) retry(ctx context.Context, cutoff int64, fn func(context.Context) error) error { |
| 779 | _, now, err := b.cfg.Clock.Now() |
| 780 | if err != nil { |
| 781 | return err |
| 782 | } |
| 783 | slack := time.Second |
| 784 | if cutoff != 0 { |
| 785 | slack = time.Duration(cutoff - now) |
| 786 | if slack <= 0 { |
| 787 | return ErrLost |
| 788 | } |
| 789 | } |
| 790 | bounded, cancel := context.WithTimeout(ctx, slack) |
| 791 | defer cancel() |
| 792 | delay := 25 * time.Millisecond |
| 793 | for { |
| 794 | err = fn(bounded) |
| 795 | if err == nil || errors.Is(err, ErrState) || errors.Is(err, ErrInactive) || errors.Is(err, ErrLost) || errors.Is(err, ErrUnsupported) || errors.Is(err, os.ErrPermission) || errors.Is(err, os.ErrInvalid) || errors.Is(err, os.ErrNotExist) || errors.Is(err, release.ErrCASConflict) { |
| 796 | return err |
| 797 | } |
| 798 | _, now, clockErr := b.cfg.Clock.Now() |
| 799 | if clockErr != nil || cutoff != 0 && now >= cutoff { |
| 800 | return errors.Join(err, clockErr, ErrLost) |
| 801 | } |
| 802 | timer := time.NewTimer(delay) |
| 803 | select { |
| 804 | case <-bounded.Done(): |
| 805 | timer.Stop() |
| 806 | return errors.Join(err, bounded.Err()) |
| 807 | case <-timer.C: |
| 808 | } |
| 809 | if delay < 500*time.Millisecond { |
| 810 | delay *= 2 |
| 811 | } |
| 812 | } |
| 813 | } |
| 814 | |
amadan