package hostexec

import (
	"bytes"
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"os"
	"strings"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/rootparticipation"
	"github.com/carlosframework/platform/internal/rootstate"
)

type ExecutionRequest = rootparticipation.ExecutionRequest
type ExecutionAuthority = rootparticipation.ExecutionAuthority
type ExecutionResponse = rootparticipation.ExecutionResponse
type ExecutionReplay = rootparticipation.ExecutionReplay
type ExecutionRecord = rootparticipation.ExecutionRecord

type RootBrokerConfig struct {
	Recovered        func(context.Context) error
	Transition       func(context.Context, ExecutionRecord, ExecutionRequest, ExecutionAuthority) error
	State            rootstate.Store
	Remote           release.Store
	Clock            Clock
	BootID, Identity string
	Authority        func(context.Context, ExecutionRequest, ControllerProcess) (ExecutionAuthority, error)
	Alive            func(context.Context, ControllerProcess) error
	Cleanup          func(context.Context, ExecutionRecord) error
	Settle           func(context.Context, ExecutionRecord) error
	Exclude          func(context.Context, func() error) error
}

type RootBroker struct {
	cfg         RootBrokerConfig
	mu          sync.Mutex
	workers     sync.WaitGroup
	stopping    bool
	disposed    bool
	joined      chan struct{}
	closeGate   chan struct{}
	lifetimeCtx context.Context
	cancel      context.CancelFunc
	closeErr    error
	denied      map[string]bool
	closed      bool
	calls       map[string]chan struct{}
	lifetime    map[string]func()
	live        map[string]func()
	retryAfter  map[string]int64
	retryDelay  map[string]time.Duration
}

func NewRootBroker(cfg RootBrokerConfig) (*RootBroker, error) {
	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 {
		return nil, ErrState
	}
	lifetimeCtx, cancel := context.WithCancel(context.Background())
	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
}

func (b *RootBroker) enter(ctx context.Context) (context.Context, func(), error) {
	b.mu.Lock()
	if b.stopping {
		b.mu.Unlock()
		return nil, nil, ErrLost
	}
	b.workers.Add(1)
	b.mu.Unlock()
	work, cancel := context.WithCancel(ctx)
	stop := context.AfterFunc(b.lifetimeCtx, cancel)
	return work, func() { stop(); cancel(); b.workers.Done() }, nil
}

func executionDigest(q ExecutionRequest, p ControllerProcess) string {
	b, _ := json.Marshal(struct {
		Request ExecutionRequest
		Peer    ControllerProcess
	}{q, p})
	s := sha256.Sum256(b)
	return hex.EncodeToString(s[:])
}

func (b *RootBroker) read(host string) (ExecutionRecord, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	raw, err := b.cfg.State.Read(host + ".json")
	if err != nil {
		return ExecutionRecord{}, err
	}
	var r ExecutionRecord
	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 {
		return ExecutionRecord{}, ErrState
	}
	switch r.Status {
	case "pending", "active", "lost", "settled":
	default:
		return ExecutionRecord{}, ErrState
	}
	return r, nil
}

func (b *RootBroker) write(r ExecutionRecord) error {
	b.mu.Lock()
	defer b.mu.Unlock()
	if r.Status == "active" && (b.closed || b.denied[r.Request.Host]) {
		return ErrLost
	}
	if release := b.live[r.Request.Host]; release != nil {
		release()
		delete(b.live, r.Request.Host)
	}
	raw, err := json.Marshal(r)
	if err != nil {
		return err
	}
	if err := b.cfg.State.Write(r.Request.Host+".json", raw); err != nil {
		return err
	}
	if r.Status == "active" {
		release, err := rootstate.HoldLive(b.cfg.State, r.Request.Host+".live")
		if err != nil {
			return err
		}
		b.live[r.Request.Host] = release
	}
	return nil
}

func executionResponse(r ExecutionRecord) ExecutionResponse {
	q := r.Request
	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}
}

func (b *RootBroker) active(ctx context.Context, r ExecutionRecord) error {
	b.mu.Lock()
	denied := b.closed || b.denied[r.Request.Host]
	b.mu.Unlock()
	if denied || r.Status != "active" || r.Broker != b.cfg.Identity || r.Peer.KernelBootID != b.cfg.BootID {
		return ErrLost
	}
	_, now, err := b.cfg.Clock.Now()
	if err != nil {
		return err
	}
	if now >= r.SafeUntil {
		return ErrLost
	}
	return b.cfg.Alive(ctx, r.Peer)
}

func sameExecution(q ExecutionRequest, r ExecutionRecord) bool {
	p := r.Request
	return q.Host == p.Host && q.Incarnation == p.Incarnation && q.Stamp == p.Stamp && q.Session == p.Session && q.Op == p.Op
}

func (b *RootBroker) deny(ctx context.Context, r ExecutionRecord, cause error) error {
	b.mu.Lock()
	raw, readErr := b.cfg.State.Read(r.Request.Host + ".json")
	var current ExecutionRecord
	if readErr == nil && decode(raw, &current) == nil && ((current.Token != r.Token && current.Token != r.TransitionFrom) || current.Status == "settled") {
		b.mu.Unlock()
		return cause
	}
	if readErr == nil && current.Token == r.Token && current.RequestStart > r.RequestStart {
		r = current
	}
	b.denied[r.Request.Host] = true
	if release := b.live[r.Request.Host]; release != nil {
		release()
		delete(b.live, r.Request.Host)
	}
	r.Status = "lost"
	raw, persist := json.Marshal(r)
	if persist == nil {
		persist = b.cfg.State.Write(r.Request.Host+".json", raw)
	}
	if persist != nil {
		b.closed = true
		for host, release := range b.live {
			release()
			delete(b.live, host)
		}
	}
	b.mu.Unlock()
	cleanup, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
	defer cancel()
	clean := b.cfg.Cleanup(cleanup, r)
	return errors.Join(cause, persist, clean)
}

func (b *RootBroker) Execute(ctx context.Context, peer ControllerProcess, q ExecutionRequest) (ExecutionResponse, error) {
	if !q.Valid() || peer.PID <= 0 || !validID(peer.Start) || peer.KernelBootID != b.cfg.BootID {
		return ExecutionResponse{}, ErrState
	}
	ctx, leave, enterErr := b.enter(ctx)
	if enterErr != nil {
		return ExecutionResponse{}, enterErr
	}
	defer leave()
	if q.Operation != "executionSettle" {
		if err := b.cfg.Alive(ctx, peer); err != nil {
			return ExecutionResponse{}, err
		}
	}
	b.mu.Lock()
	gate := b.calls[q.Host]
	if gate == nil {
		if len(b.calls) >= 4096 {
			b.mu.Unlock()
			return ExecutionResponse{}, ErrState
		}
		gate = make(chan struct{}, 1)
		b.calls[q.Host] = gate
	}
	b.mu.Unlock()
	select {
	case gate <- struct{}{}:
	case <-ctx.Done():
		return ExecutionResponse{}, ctx.Err()
	}
	defer func() { <-gate }()
	b.mu.Lock()
	held := b.lifetime[q.Host] != nil
	b.mu.Unlock()
	if !held {
		unlock, lockErr := b.cfg.State.Lock(ctx, q.Host+".lock")
		if lockErr != nil {
			return ExecutionResponse{}, lockErr
		}
		b.mu.Lock()
		b.lifetime[q.Host] = unlock
		b.mu.Unlock()
	}
	if err := ctx.Err(); err != nil {
		return ExecutionResponse{}, err
	}
	// Settlement verifies retained obligations even after current admission ends.
	var authority ExecutionAuthority
	var err error
	if q.Operation != "executionSettle" {
		cutoff := int64(0)
		if prior, readErr := b.read(q.Host); readErr == nil && prior.Status == "active" {
			cutoff = prior.SafeUntil
		}
		err = b.retry(ctx, cutoff, func(call context.Context) error { var e error; authority, e = b.cfg.Authority(call, q, peer); return e })
		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 == "") {
			err = ErrState
		}
		if err != nil {
			if prior, readErr := b.read(q.Host); readErr == nil && prior.Peer == peer && (q.Token == prior.Token || q.ID == prior.Request.ID) && prior.Status != "settled" {
				return ExecutionResponse{}, b.deny(ctx, prior, err)
			}
			return ExecutionResponse{}, err
		}
		if q.Operation == "executionView" && q.Token == "" {
			return b.servingView(ctx, peer, q, authority)
		}
		if authority.Kind == "serving" {
			return ExecutionResponse{}, ErrInactive
		}
	}
	r, err := b.read(q.Host)
	missing := errors.Is(err, os.ErrNotExist)
	if err != nil && !missing {
		return ExecutionResponse{}, err
	}
	digest := executionDigest(q, peer)
	if !missing {
		if replay, ok := r.Replay[q.ID]; ok {
			if replay.Digest != digest {
				return ExecutionResponse{}, ErrState
			}
			if q.Operation == "executionSettle" && replay.Response.Status == "settled" {
				return replay.Response, nil
			}
			if !sameExecution(q, r) || replay.Response.Token != r.Token {
				return ExecutionResponse{}, ErrLost
			}
			if err = b.active(ctx, r); err != nil {
				return ExecutionResponse{}, b.deny(ctx, r, err)
			}
			if r.Peer != peer || r.Authority != authority {
				return ExecutionResponse{}, b.deny(ctx, r, ErrLost)
			}
			return replay.Response, nil
		}
	}
	if q.Operation == "executionBegin" {
		b.mu.Lock()
		closed := b.closed
		b.mu.Unlock()
		if closed || (!missing && (r.Status != "settled" || sameExecution(q, r))) {
			return ExecutionResponse{}, ErrLost
		}
		var token [32]byte
		if _, err = rand.Read(token[:]); err != nil {
			return ExecutionResponse{}, err
		}
		history := make(map[string]ExecutionReplay)
		for id, replay := range r.Replay {
			if replay.Response.Status == "settled" && replay.Response.Token == r.Token {
				history[id] = replay
			}
		}
		b.mu.Lock()
		delete(b.denied, q.Host)
		delete(b.retryAfter, q.Host)
		delete(b.retryDelay, q.Host)
		b.mu.Unlock()
		r = ExecutionRecord{Schema: 1, Request: q, Peer: peer, Authority: authority, Broker: b.cfg.Identity, Token: hex.EncodeToString(token[:]), Status: "pending", Replay: history}
		wall, start, e := b.cfg.Clock.Now()
		if e != nil {
			return ExecutionResponse{}, e
		}
		r.PriorSafeUntil = r.SafeUntil
		r.RequestStart = start
		r.SafeUntil = start + int64(TTL-Margin)
		r.Expires = wall.Add(TTL)
		r.Replay[q.ID] = ExecutionReplay{Digest: digest}
		if err = b.cfg.Exclude(ctx, func() error { return b.write(r) }); err != nil {
			return ExecutionResponse{}, err
		}
		return b.acquire(ctx, peer, q, r, digest, true)
	}
	if missing || r.Peer != peer || q.Token != r.Token {
		return ExecutionResponse{}, ErrLost
	}
	if !sameExecution(q, r) {
		if q.Operation != "executionRenew" || b.cfg.Transition == nil || q.Host != r.Request.Host || q.Op != r.Request.Op || q.Incarnation != r.Request.Incarnation {
			return ExecutionResponse{}, ErrLost
		}
		if err = b.active(ctx, r); err != nil {
			return ExecutionResponse{}, b.deny(ctx, r, err)
		}
		if err = b.cfg.Transition(ctx, r, q, authority); err != nil {
			return ExecutionResponse{}, b.deny(ctx, r, err)
		}
		var token [32]byte
		if _, err = rand.Read(token[:]); err != nil {
			return ExecutionResponse{}, err
		}
		r.Request = q
		r.TransitionFrom = r.Token
		r.Token = hex.EncodeToString(token[:])
		r.Authority = authority
		return b.acquire(ctx, peer, q, r, digest, false)
	}
	if q.Operation == "executionSettle" {
		return b.settle(ctx, q, r, digest)
	}
	if err = b.active(ctx, r); err != nil {
		return ExecutionResponse{}, b.deny(ctx, r, err)
	}
	if authority != r.Authority {
		if !executionInstallationBinding(r.Authority, authority) {
			if (q.Operation != "executionRenew" && q.Operation != "executionView") || b.cfg.Transition == nil || b.cfg.Transition(ctx, r, q, authority) != nil {
				return ExecutionResponse{}, b.deny(ctx, r, ErrLost)
			}
			r.Authority = authority
			if q.Operation == "executionView" {
				if err = b.write(r); err != nil {
					return ExecutionResponse{}, b.deny(ctx, r, err)
				}
				return executionResponse(r), nil
			}
			return b.acquire(ctx, peer, q, r, digest, false)
		}
		r.Authority = authority
		if err = b.write(r); err != nil {
			return ExecutionResponse{}, b.deny(ctx, r, err)
		}
	}
	if q.Operation == "executionView" {
		return executionResponse(r), nil
	}
	return b.acquire(ctx, peer, q, r, digest, false)
}

func (b *RootBroker) acquire(ctx context.Context, peer ControllerProcess, q ExecutionRequest, r ExecutionRecord, digest string, begin bool) (ExecutionResponse, error) {
	key, err := Key(q.Host, r.Authority.Box)
	if err != nil {
		return ExecutionResponse{}, err
	}
	etag := r.LeaseETag
	if begin {
		raw, existing, e := b.cfg.Remote.Get(ctx, key)
		if e == nil {
			var remote Record
			wall, _, clockErr := b.cfg.Clock.Now()
			if decode(raw, &remote) != nil || remote.Expires.IsZero() || !wall.After(remote.Expires) || existing == "" || clockErr != nil {
				return ExecutionResponse{}, b.deny(ctx, r, ErrLost)
			}
			etag = existing
		} else if !errors.Is(e, os.ErrNotExist) {
			return ExecutionResponse{}, b.deny(ctx, r, e)
		}
	} else {
		if r.Token == q.Token {
			var token [32]byte
			if _, err := rand.Read(token[:]); err != nil {
				return ExecutionResponse{}, err
			}
			r.TransitionFrom, r.Token = r.Token, hex.EncodeToString(token[:])
		}
		priorID := ""
		var prior ExecutionReplay
		for id, replay := range r.Replay {
			if replay.Response.Status == "settled" && (priorID == "" || replay.Response.Expires.After(prior.Response.Expires) || replay.Response.Expires.Equal(prior.Response.Expires) && id > priorID) {
				priorID, prior = id, replay
			}
		}
		r.Replay = make(map[string]ExecutionReplay)
		if priorID != "" {
			r.Replay[priorID] = prior
		}
		wall, start, e := b.cfg.Clock.Now()
		if e != nil {
			return ExecutionResponse{}, b.deny(ctx, r, e)
		}
		r.PriorSafeUntil = r.SafeUntil
		r.RequestStart = start
		r.SafeUntil = start + int64(TTL-Margin)
		r.Expires = wall.Add(TTL)
		r.Status = "pending"
		r.Replay[q.ID] = ExecutionReplay{Digest: digest}
		if e = b.cfg.Exclude(ctx, func() error { return b.write(r) }); e != nil {
			return ExecutionResponse{}, b.deny(ctx, r, e)
		}
	}
	raw, _ := json.Marshal(Record{Op: q.Op, Expires: r.Expires})
	var next string
	cutoff := r.SafeUntil
	if r.PriorSafeUntil > 0 && r.PriorSafeUntil < cutoff {
		cutoff = r.PriorSafeUntil
	}
	uncertain := false
	err = b.retry(ctx, cutoff, func(call context.Context) error {
		if uncertain {
			found, current, e := b.cfg.Remote.Get(call, key)
			if e == nil && bytes.Equal(found, raw) && current != "" {
				next = current
				return nil
			}
			if e != nil && !errors.Is(e, os.ErrNotExist) {
				return e
			}
			if e == nil && current != etag {
				return ErrLost
			}
		}
		var e error
		next, e = b.cfg.Remote.Put(call, key, raw, etag)
		if e != nil && !errors.Is(e, release.ErrCASConflict) {
			uncertain = true
		}
		return e
	})
	if err != nil || next == "" {
		return ExecutionResponse{}, b.deny(ctx, r, errors.Join(err, ErrLost))
	}
	r.LeaseETag = next
	var authority ExecutionAuthority
	err = b.retry(ctx, cutoff, func(call context.Context) error { var e error; authority, e = b.cfg.Authority(call, q, peer); return e })
	if err != nil || authority != r.Authority {
		return ExecutionResponse{}, b.deny(ctx, r, errors.Join(err, ErrLost))
	}
	_, now, err := b.cfg.Clock.Now()
	if err != nil || now >= r.SafeUntil || (!begin && r.PriorSafeUntil > 0 && now >= r.PriorSafeUntil) || ctx.Err() != nil {
		return ExecutionResponse{}, b.deny(ctx, r, errors.Join(err, ErrLost))
	}
	if err = b.cfg.Alive(ctx, peer); err != nil {
		return ExecutionResponse{}, b.deny(ctx, r, err)
	}
	r.PriorSafeUntil = 0
	r.Status = "active"
	out := executionResponse(r)
	r.Replay[q.ID] = ExecutionReplay{Digest: digest, Response: out}
	if err = b.write(r); err != nil {
		return ExecutionResponse{}, b.deny(ctx, r, err)
	}
	if err = b.active(ctx, r); err != nil {
		return ExecutionResponse{}, b.deny(ctx, r, err)
	}
	return out, nil
}

func (b *RootBroker) settle(ctx context.Context, q ExecutionRequest, r ExecutionRecord, digest string) (ExecutionResponse, error) {
	if r.Status == "settled" {
		return executionResponse(r), nil
	}
	if err := b.deny(ctx, r, nil); err != nil {
		return ExecutionResponse{}, err
	}
	r.Status = "lost"
	key, err := Key(q.Host, r.Authority.Box)
	if err != nil {
		return ExecutionResponse{}, err
	}
	deletionUncertain := r.LeaseETag == ""
	if !deletionUncertain {
		err = release.DeleteIfMatch(ctx, b.cfg.Remote, key, r.LeaseETag)
		if errors.Is(err, release.ErrCASConflict) {
			deletionUncertain = true
		} else if err != nil && !errors.Is(err, os.ErrNotExist) {
			return ExecutionResponse{}, err
		}
	}
	if deletionUncertain {
		raw, currentETag, readErr := b.cfg.Remote.Get(ctx, key)
		wall, _, clockErr := b.cfg.Clock.Now()
		if clockErr != nil {
			return ExecutionResponse{}, clockErr
		}
		if !wall.After(r.Expires) {
			return ExecutionResponse{}, ErrLost
		}
		if readErr == nil {
			var remote Record
			if decode(raw, &remote) != nil || remote.Expires.IsZero() || !wall.After(remote.Expires) || currentETag == "" {
				return ExecutionResponse{}, ErrLost
			}
			if err := release.DeleteIfMatch(ctx, b.cfg.Remote, key, currentETag); err != nil {
				return ExecutionResponse{}, err
			}
		} else if !errors.Is(readErr, os.ErrNotExist) {
			return ExecutionResponse{}, readErr
		}
	}
	if err = b.cfg.Settle(ctx, r); err != nil {
		return ExecutionResponse{}, err
	}
	r.Status = "settled"
	out := executionResponse(r)
	r.Replay = make(map[string]ExecutionReplay)
	r.Replay[q.ID] = ExecutionReplay{Digest: digest, Response: out}
	if err = b.write(r); err != nil {
		return ExecutionResponse{}, err
	}
	return out, nil
}

func (b *RootBroker) Monitor(ctx context.Context) error {
	ctx, leave, err := b.enter(ctx)
	if err != nil {
		return err
	}
	defer leave()
	names, err := b.cfg.State.Names()
	if err != nil {
		return err
	}
	var errs []error
	for _, name := range names {
		if name == "census.json" || name == "update.json" || !strings.HasSuffix(name, ".json") {
			continue
		}
		host := strings.TrimSuffix(name, ".json")
		b.mu.Lock()
		owned := b.lifetime[host] != nil
		b.mu.Unlock()
		if !owned {
			continue
		}
		r, err := b.read(host)
		if err != nil {
			errs = append(errs, err)
			continue
		}
		if r.Status == "settled" {
			continue
		}
		if r.Status == "lost" {
			_, now, clockErr := b.cfg.Clock.Now()
			if clockErr != nil {
				errs = append(errs, clockErr)
				continue
			}
			b.mu.Lock()
			next := b.retryAfter[host]
			b.mu.Unlock()
			if now < next {
				continue
			}
			err := b.recoverSettlement(ctx, host)
			b.mu.Lock()
			if err != nil {
				delay := b.retryDelay[host] * 2
				if delay < time.Second {
					delay = time.Second
				}
				if delay > 30*time.Second {
					delay = 30 * time.Second
				}
				b.retryDelay[host], b.retryAfter[host] = delay, now+int64(delay)
			} else {
				delete(b.retryDelay, host)
				delete(b.retryAfter, host)
			}
			b.mu.Unlock()
			errs = append(errs, err)
			continue
		}
		_, now, clockErr := b.cfg.Clock.Now()
		bounded, cancel := context.WithTimeout(ctx, time.Second)
		aliveErr := b.cfg.Alive(bounded, r.Peer)
		cancel()
		cutoff := r.SafeUntil
		if r.Status == "pending" && r.PriorSafeUntil > 0 && r.PriorSafeUntil < cutoff {
			cutoff = r.PriorSafeUntil
		}
		if clockErr != nil || aliveErr != nil || now >= cutoff || r.Broker != b.cfg.Identity || r.Peer.KernelBootID != b.cfg.BootID {
			errs = append(errs, b.deny(ctx, r, ErrLost))
		}
	}
	if b.cfg.Recovered != nil {
		errs = append(errs, b.cfg.Recovered(ctx))
	}
	return errors.Join(errs...)
}

func executionInstallationBinding(old, next ExecutionAuthority) bool {
	if !old.ParticipationOnly || (old.Installation != "" && (old.Installation != next.Installation || next.ParticipationOnly)) || next.Installation == "" {
		return false
	}
	old.Installation = next.Installation
	old.ParticipationOnly = next.ParticipationOnly
	return old == next
}

func (b *RootBroker) Close(ctx context.Context) error {
	b.mu.Lock()
	b.closed = true
	if !b.stopping {
		b.stopping = true
		b.cancel()
		for host, unlock := range b.live {
			unlock()
			delete(b.live, host)
		}
		b.joined = make(chan struct{})
		go func() { b.workers.Wait(); close(b.joined) }()
	}
	joined := b.joined
	b.mu.Unlock()
	select {
	case <-joined:
	case <-ctx.Done():
		return ctx.Err()
	}
	select {
	case b.closeGate <- struct{}{}:
	case <-ctx.Done():
		return ctx.Err()
	}
	defer func() { <-b.closeGate }()
	if b.disposed {
		return b.closeErr
	}
	names, err := b.cfg.State.Names()
	errs := []error{err}
	for _, name := range names {
		if name == "census.json" || name == "update.json" || !strings.HasSuffix(name, ".json") {
			continue
		}
		r, e := b.read(strings.TrimSuffix(name, ".json"))
		if e != nil {
			errs = append(errs, e)
			continue
		}
		if r.Broker == b.cfg.Identity && r.Status != "settled" {
			errs = append(errs, b.deny(ctx, r, nil))
		}
	}
	b.mu.Lock()
	for host, unlock := range b.lifetime {
		unlock()
		delete(b.lifetime, host)
	}
	b.mu.Unlock()
	b.disposed = true
	b.closeErr = errors.Join(errs...)
	return b.closeErr
}

func (b *RootBroker) Recover(ctx context.Context) error {
	ctx, leave, err := b.enter(ctx)
	if err != nil {
		return err
	}
	defer leave()
	names, err := b.cfg.State.Names()
	if err != nil {
		return err
	}
	var errs []error
	for _, name := range names {
		if name == "census.json" || name == "update.json" || !strings.HasSuffix(name, ".json") {
			continue
		}
		host := strings.TrimSuffix(name, ".json")
		b.mu.Lock()
		owned := b.lifetime[host] != nil
		b.mu.Unlock()
		if !owned {
			unlock, e := b.cfg.State.Lock(ctx, host+".lock")
			if e != nil {
				return e
			}
			b.mu.Lock()
			b.lifetime[host] = unlock
			b.mu.Unlock()
		}
		r, e := b.read(host)
		if e != nil {
			errs = append(errs, e)
			continue
		}
		if r.Status != "settled" {
			errs = append(errs, b.recoverSettlement(ctx, host))
		}
	}
	if b.cfg.Recovered != nil {
		errs = append(errs, b.cfg.Recovered(ctx))
	}
	return errors.Join(errs...)
}

func (b *RootBroker) recoverSettlement(ctx context.Context, host string) error {
	b.mu.Lock()
	gate := b.calls[host]
	if gate == nil {
		gate = make(chan struct{}, 1)
		b.calls[host] = gate
	}
	b.mu.Unlock()
	select {
	case gate <- struct{}{}:
	default:
		return nil
	}
	defer func() { <-gate }()
	r, err := b.read(host)
	if err != nil || r.Status == "settled" {
		return err
	}
	q := r.Request
	q.Operation, q.Token, q.ID = "executionSettle", r.Token, "recovery-"+r.Token
	_, err = b.settle(ctx, q, r, executionDigest(q, r.Peer))
	return err
}

func (b *RootBroker) Run(ctx context.Context) {
	ctx, leave, err := b.enter(ctx)
	if err != nil {
		return
	}
	defer leave()
	for {
		select {
		case <-ctx.Done():
			return
		case <-b.lifetimeCtx.Done():
			return
		case <-b.cfg.Clock.After(100 * time.Millisecond):
		}
		_ = b.Monitor(ctx)
	}
}

func (b *RootBroker) retry(ctx context.Context, cutoff int64, fn func(context.Context) error) error {
	_, now, err := b.cfg.Clock.Now()
	if err != nil {
		return err
	}
	slack := time.Second
	if cutoff != 0 {
		slack = time.Duration(cutoff - now)
		if slack <= 0 {
			return ErrLost
		}
	}
	bounded, cancel := context.WithTimeout(ctx, slack)
	defer cancel()
	delay := 25 * time.Millisecond
	for {
		err = fn(bounded)
		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) {
			return err
		}
		_, now, clockErr := b.cfg.Clock.Now()
		if clockErr != nil || cutoff != 0 && now >= cutoff {
			return errors.Join(err, clockErr, ErrLost)
		}
		timer := time.NewTimer(delay)
		select {
		case <-bounded.Done():
			timer.Stop()
			return errors.Join(err, bounded.Err())
		case <-timer.C:
		}
		if delay < 500*time.Millisecond {
			delay *= 2
		}
	}
}