package hostlife

// ---------- the application ----------

// traffic: every running process of the host does its own work, whether or
// not a request arrived: background workers and descendants included.
func (w *world) traffic(b *box) {
	if b.Destroyed || b.Frozen {
		return
	}
	for _, p := range b.Procs {
		if w.rng.Intn(3) != 0 {
			continue
		}
		w.nextW++
		wr := w.nextW
		if b.DB != nil && b.DB.Inc == p.Inc {
			w.commit(b.DB, wr)
		} else {
			w.txs[wr] = tx{W: wr} // into a purged or foreign database: nowhere
		}
		// acknowledged to the client or the outside world either way: a
		// write into a purged or foreign database is a lost write. The
		// acknowledgement is the process's session's (rev 7.13, R12-1): the
		// database under it may be gone (seed 899 / 3,000)
		w.acks = append(w.acks, ack{W: wr, Inc: p.Inc, Box: b.ID, Session: p.Session, Gen: p.Gen, At: w.now, Kind: "write"})
		w.writeInc[wr] = p.Inc
		if p.Main && w.rng.Intn(4) == 0 && b.Row != nil && b.Row.Inc == p.Inc {
			w.nextW++
			b.Timers = append(b.Timers, w.nextW)
			w.acks = append(w.acks, ack{W: w.nextW, Inc: p.Inc, Box: b.ID, Session: p.Session, Gen: p.Gen, At: w.now, Kind: "timer"})
		}
	}
}

// idleStop: the activator's ordinary idle stop (main only; a descendant
// keeps running, as today).
func (w *world) idleStop(b *box) {
	if !w.agentUp(b) || !hasMain(b) {
		return
	}
	w.replicate(b)
	// the unit stop kills the cgroup: the main process and every descendant
	// placed in it; an escaped descendant survives, and while it lives the
	// session cannot certify (rev 7.10)
	var keep []*proc
	for _, p := range b.Procs {
		if !p.Main && !p.Strict {
			keep = append(keep, p)
		}
	}
	b.Procs = keep
	if b.open != nil {
		// the session's start is still finishing its PUTs: the agent's
		// replicator opens when they land and ships what the app wrote,
		// then closes (rev 7.11 §7.14); no certificate now
		w.stats["idle stop before replication opened"]++
		return
	}
	// the clean close (claim rev 6.9 §6.4a): drain, then the certificate
	if b.DB != nil {
		b.DB.Closing = true
		w.closeStep(b)
	}
}

// wake: a request reaches this box's edge for the host.
func (w *world) wake(b *box) {
	if !w.agentUp(b) || b.Row == nil || b.Row.Status != "ready" {
		return
	}
	// the edge routes on the row (edge.go:595): a ready row receives the
	// public request whether or not the instance already runs. INV-5: no
	// public delivery to a landing target before the commit (R3-5)
	if c := w.claim; c != nil && c.Op.Kind == "landing" && !c.Op.Committed && c.Op.Tgt == b.ID {
		w.violate("INV-5: a public request reached the landing target, box %d, before the commit of %s", b.ID, c.Op.ID)
		return
	}
	if hasMain(b) {
		return
	}
	if b.Old {
		if w.rec == nil {
			return
		}
		c := &claim{Holder: &holder{Inc: w.rec.Inc, App: w.rec.App, Legacy: true}}
		w.spawn(b, c, "", false)
		return
	}
	c, ok := w.admissionClaim(b)
	if !ok || !w.admits(b, c, "") {
		return
	}
	if !w.completedHandover(b, c) {
		return // N6: the first serving wake after a completion hands over from the probe's receipt
	}
	if b.DB == nil && on("quiet-before-ack") && w.now < c.Quiet[b.Region] {
		return // a first install into this region waits out quiet_after (claim §6.2a)
	}
	w.spawn(b, c, c.Holder.Gen, false)
}

// ---------- reconcile ----------

func (w *world) retireRow(b *box) {
	if b.Row == nil {
		return
	}
	if !b.Old && !w.evidence(b, "retire") {
		return
	}
	if b.Old {
		var keep []*proc
		for _, p := range b.Procs {
			if !p.Main {
				keep = append(keep, p)
			}
		}
		b.Procs = keep
	}
	b.Row = nil
	b.Config = 0 // reapConfigEnv
	// the schedule converger prunes timers of a host with no row
	b.Timers, b.History, b.Attempt = nil, nil, 0
}

func (w *world) ownsLocalObligation(c *claim, b *box) bool {
	for _, o := range c.Obl {
		if o.Kind == "park-local" && o.Box == b.ID {
			return true
		}
	}
	return false
}

func (w *world) reconcile(b *box) {
	if !w.agentUp(b) {
		return
	}
	if b.Old {
		if w.rec != nil && w.rec.Region == b.Region {
			if b.Row == nil || b.Row.Inc != w.rec.Inc {
				b.Row = &row{Inc: w.rec.Inc, App: w.rec.App, Status: "ready"}
			}
		} else if b.Row != nil {
			w.retireRow(b)
		}
		return
	}
	c, ok := w.admissionClaim(b) // reconcile's keep-or-create is an admission decision (C4)
	if !ok {
		return
	}
	if c == nil {
		if w.claim == nil {
			w.adopt()
		}
		return
	}
	w.reconcileClaimed(b, c)
}

// adopt is the one adoption routine every actor uses (claim §9): a
// create-only claim for an unclaimed legacy record, with discovery.
func (w *world) adopt() bool {
	if w.claim != nil || w.rec == nil {
		return false
	}
	eb := w.adoptionOwner(w.rec.Region) // exactly one proven owner (claim P5-D2)
	if eb == nil {
		return false
	}
	// adoption writes holder.stamp (claim rev 6.5 §2). The pre-claim writer
	// is the legacy identity (claim rev 6.24, R16-3): session 0 with
	// generation "" (read as @root), the pre-release process on the
	// executing box its writer_box; the first cold admission's writes_lost
	// names it (seed 94288 / 100,000 named box 0 and left the legacy writes
	// undeclared), and a lost-target return declares it in
	// additional_writes_lost. Rev 7.13 to 7.16 gave it session 1, which hid
	// the session-zero case from E
	n := &claim{Ver: 1, Holder: &holder{Inc: w.rec.Inc, App: w.rec.App, Legacy: true, Location: locationFor(w.rec.Inc, true), Stamp: 1, Session: 0, WriterBox: eb.ID},
		Op: opState{Kind: "serving", Exec: eb.ID, Parts: []int{eb.ID}}, Regions: []int{w.rec.Region}, Boxes: []int{eb.ID}, Complete: false}
	if w.stampBox == nil {
		w.stampBox = map[int]int{}
	}
	w.stampBox[1] = eb.ID // the adoption's stamp names the serving box's installation (claim rev 6.5 §2)
	if on("legacy-discovery") {
		// every other region parks whatever it holds: it cannot be the
		// live lineage, because the record names another region
		for r := 0; r < nRegions; r++ {
			if r == w.rec.Region {
				continue
			}
			for _, p := range []string{"instances/" + host + "/", "moves/" + host + "/"} {
				n.Obl = append(n.Obl, obligation{ID: w.newID("ob"), Op: "adopt", Kind: "park-replica", Region: r, Box: -1, Prefix: p, Ret: "legacy"})
			}
		}
		// and local state on every box but the record region's serving box
		// (claim rev 4 §9, the J2 N5 note): it cannot be the live instance's
		for _, b := range w.liveBoxes() {
			if b.ID != eb.ID {
				n.Obl = append(n.Obl, obligation{ID: w.newID("ob"), Op: "adopt", Kind: "park-local", Box: b.ID, Region: -1, Ret: "legacy"})
			}
		}
	}
	w.claim = n
	w.logf("adopted legacy host inc %d on box %d", w.rec.Inc, eb.ID)
	return true
}

func (w *world) reconcileClaimed(b *box, c *claim) {
	h := c.Holder
	if h == nil || w.rec == nil || w.rec.Inc != h.Inc {
		if b.Row != nil && !w.ownsLocalObligation(c, b) {
			w.retireRow(b)
		}
		return
	}
	if b.Row != nil && b.Row.Inc != h.Inc {
		w.retireRow(b) // an orphan row (claim C1)
		return
	}
	switch c.Op.Kind {
	case "serving", "creating", "pushing":
		if c.Op.Exec == b.ID {
			if b.Row != nil && b.Row.Status != "ready" && c.Op.Kind == "serving" && on("stale-hold-release") && !w.completionHolds(c, b) {
				// a hold whose operation has ended (the claim is serving here)
				b.Row.Status, b.Row.HoldOp = "ready", ""
			}
			if b.Row == nil && c.Op.Kind == "serving" {
				b.Row = &row{Inc: h.Inc, App: h.App, Status: "ready"}
				w.initRecord(b, c) // the local record from holder.stamp, on this fresh read (claim rev 6.5 §6.4)
			}
		} else if w.catchUpHold(b, c) {
			// a delayed source's row, held for the completed op until its
			// park-local retires it (finding 3, ruling 85)
		} else if b.Row != nil && !(on("participant-rows") && w.ownsLocalObligation(c, b)) {
			w.retireRow(b)
		}
	default:
		if c.Op.Kind == "relocating" && b.ID == c.Op.Src && b.Row != nil && b.Row.Status == "ready" && c.Op.Release != nil {
			// §5.3, L12 (rule reloc-row-cases): the former executor's row is
			// held once its admission fence closed, never ready, until its
			// park-local retires it. The mutation reopens it as a stale hold
			if on("reloc-row-cases") {
				b.Row.Status, b.Row.HoldOp = "relocating", c.Op.ID
			}
		}
		if w.isParticipant(c, b) {
			if !on("participant-rows") && w.rec.Region != b.Region && b.Row != nil {
				w.retireRow(b) // region mismatch retires a participant's row
			}
			return
		}
		if b.Row != nil && !w.ownsLocalObligation(c, b) {
			w.retireRow(b)
		}
	}
}

// ---------- background passes with work in flight (J-N8) ----------

func (w *world) passPlan(b *box, kind string) {
	if !w.agentUp(b) || b.Row == nil || b.Row.Status != "ready" {
		return
	}
	if kind == "config" && !b.HasKey {
		return
	}
	b.plans = append(b.plans, &plan{Kind: kind, Inc: b.Row.Inc, App: b.Row.App, Ver: w.cfgVer[b.Row.App], At: w.now + w.pauseLen(), Stamp: b.stampOf(b.Row.Inc)})
}

// passExec runs one independent process's write. It needs the box running,
// not the agent. Under the host guard (an exclusive flock shared by the
// agent and every root command, claim rev 4 §6.4) it reads the claim
// fresh and refuses if the incarnation, the operation or the config
// identity it selected has moved on. The mutation keeps an in-process lock
// only: the process acts on what it validated before it paused.
func (w *world) passExec(b *box) {
	if b.Destroyed || b.Frozen || len(b.plans) == 0 {
		return
	}
	i := w.rng.Intn(len(b.plans))
	p := b.plans[i]
	if p.At > w.now {
		return
	}
	b.plans = append(b.plans[:i:i], b.plans[i+1:]...)
	if on("hostguard") {
		if b.Partitioned {
			return
		}
		c := w.claim
		if c == nil || c.Holder == nil || c.Holder.Inc != p.Inc || b.Row == nil || b.Row.Inc != p.Inc {
			return
		}
		if p.Kind == "config" && p.Ver != w.cfgVer[p.App] {
			return // the selected bundle is no longer the held one
		}
	}
	// the local record (claim rev 6.5 §6.4): the writer carries the stamp
	// it selected under, admitted only if it is still the active one
	if !b.admitsStamp(p.Inc, p.Stamp) {
		return
	}
	if p.Stamp != b.stampOf(p.Inc) {
		// the oracle: a writer of a retired installation wrote host state
		w.violate("INV-6: a %s pass of a retired installation (stamp %d, active %d) wrote host state for inc %d on box %d", p.Kind, p.Stamp, b.stampOf(p.Inc), p.Inc, b.ID)
	}
	switch p.Kind {
	case "config":
		b.Config, b.CfgVer = p.Inc, p.Ver
	case "adopt":
		b.Live, b.LiveInc = p.App, p.Inc
	}
}

// ---------- timers ----------

func (w *world) runner(b *box) {
	if !w.agentUp(b) || b.Row == nil {
		return
	}
	if len(b.Timers) > 0 {
		if b.Row.Status == "ready" && hasMain(b) {
			id := b.Timers[0]
			b.Timers = b.Timers[1:]
			b.History = append(b.History, id)
			w.delivered[id] = true
		} else if b.Row.Status != "ready" && !on("timer-freeze") {
			b.Attempt++ // a refused delivery still advances bookkeeping
		}
	}
	if w.rng.Intn(4) == 0 {
		b.History = nil // the runner's own daily prune (runner.go:260)
	}
}

// ---------- self-fencing (claim §7.1) ----------

func (w *world) watchdog(b *box) {
	if b.Destroyed || b.Frozen {
		return
	}
	if !on("watchdog-independent") && !w.agentUp(b) {
		return
	}
	if b.ExecUntil > 0 && w.localNow(b) > b.ExecUntil && len(b.Procs) > 0 {
		b.Procs = nil
		w.logf("box %d self-fenced (exec lease expired)", b.ID)
	}
}

// ---------- faults ----------

func (w *world) fault() {
	if len(w.pending) > 0 && w.rng.Intn(3) == 0 {
		i := w.rng.Intn(len(w.pending))
		r := w.pending[i]
		if r.beyond && w.fail["recreate"] {
			return // the biased schedule keeps it for its worst moment
		}
		w.pending = append(w.pending[:i:i], w.pending[i+1:]...)
		w.complete(r)
		return
	}
	bs := w.liveBoxes()
	b := bs[w.rng.Intn(len(bs))]
	switch w.rng.Intn(9) {
	case 0:
		if !b.AgentDown {
			b.AgentDown = true
			var keep []*proc
			for _, p := range b.Procs {
				if b.Nested {
					keep = append(keep, p)
				}
			}
			b.Procs = keep
			b.forget()
			w.logf("box %d agent crashed", b.ID)
			// a crash that takes the disk with it (an instance replaced, a
			// volume gone): the writer's uploads may still be in flight,
			// which is the writer-epoch hazard (claim rev 6.6 §6.4a, R7-1)
			// in every operation that has this box write (R8-2), not only
			// while serving
			if c := w.claim; w.fail["disk"] && c != nil && c.Holder != nil && c.Op.Exec == b.ID && b.DB != nil && w.rng.Intn(3) == 0 {
				// the fault removes physical state only (R10-6): the
				// acknowledged, unuploaded writes are declared lost by the
				// next admission's writes_lost record, never exempted here
				b.DB = nil
				b.Procs = nil // nothing survives the disk: a descendant with no database would acknowledge writes into nowhere
				w.stats["disk-lost with the crash"]++
				w.logf("box %d lost its disk with the crash", b.ID)
			}
		} else {
			b.AgentDown = false
		}
	case 1:
		b.Procs = nil
		b.plans = nil // a reboot ends every process
		b.AgentDown, b.Frozen, b.AgentPaused = false, false, false
		b.forget()
		if w.rng.Intn(6) == 0 {
			// the purge of the local database the next admitted writer must
			// make returns an error for a while (claim rev 6.9): the writer
			// does not start
			b.PurgeFails = !b.PurgeFails
			if b.PurgeFails {
				w.stats["purge failures injected"]++
			}
		}
		if w.rng.Intn(8) == 0 && len(b.Stamps) > 0 {
			// the local stamp record is lost with the reboot: the box
			// refuses every writer until it re-initializes from a fresh
			// claim read naming it (claim rev 6.5 §6.4; rev 7.7, R6-4)
			b.Stamps = nil
			w.logf("box %d lost its local stamp record", b.ID)
			w.stats["stamp records lost"]++
		}
		if !on("floor-boot-recovery") && w.rng.Intn(3) == 0 {
			b.Old = true
			w.logf("box %d boot-recovered an old executable", b.ID)
		}
		w.logf("box %d rebooted", b.ID)
	case 2:
		if w.fail["partition"] {
			b.Partitioned = !b.Partitioned
		}
	case 3:
		if w.fail["freezetgt"] {
			// the biased schedule (rev 7.5): freeze the landing target in
			// its probe or settle window, and keep it frozen until the
			// source runs again, if a declare-lost let it
			c := w.claim
			if c != nil && c.Op.Kind == "landing" && b.ID == c.Op.Tgt && !b.Frozen && (c.Op.LandPhase == "probing" || c.Op.LandPhase == "probed") {
				b.Frozen = true
				w.logf("box %d frozen in its settle window (biased schedule)", b.ID)
				break
			}
			if b.Frozen && c != nil && (c.Op.Kind == "landing" || c.Op.Kind == "restoring") && b.ID == c.Op.Tgt {
				if s := w.boxByID(c.Op.Src); s != nil && hasMain(s) {
					b.Frozen = false
					w.logf("box %d resumed beside the recovered source (biased schedule)", b.ID)
				}
				break
			}
		}
		if w.fail["freeze"] {
			b.Frozen = !b.Frozen
		}
	case 4:
		if w.fail["pause"] {
			b.AgentPaused = !b.AgentPaused
		}
	case 5:
		c := w.claim
		committed := c != nil && (c.Op.Committed || w.rec != nil && w.rec.MoveID == c.Op.ID && w.rec.MoveState == "committed")
		if w.fail["destroy"] && c != nil && (c.Op.Kind == "landing" || c.Op.Kind == "restoring") && !committed && b.ID == c.Op.Tgt {
			b.Destroyed = true
			b.Procs = nil
			w.logf("box %d destroyed", b.ID)
		}
	case 6:
		w.console = nil // console crash: plans dropped
	case 7:
		// the serving box loses its database file while the app is idle: the
		// next wake restores cold. Writes after the last completed
		// checkpoint are lost by design (the RPO), and declared so; every
		// durable write must come back (§14.2.3).
		c := w.claim
		if w.fail["disk"] && c != nil && c.Holder != nil && c.Op.Kind == "serving" && c.Op.Exec == b.ID && b.DB != nil && !hasMain(b) && len(b.Procs) == 0 {
			// physical loss only (R10-6): the next admission declares
			b.DB = nil
			w.stats["disk-lost"]++
			w.logf("box %d lost its database file", b.ID)
		}
	default:
		w.completeDue(false)
	}
}

func (w *world) replaceDestroyed() {
	for r := 0; r < nRegions; r++ {
		for len(w.regionBoxes(r)) < w.regionMin(r) { // the region's minimum (H5): the group launches replacements
			w.nextID++
			nb := &box{ID: 100 + w.nextID, Region: r, HasKey: true, Enrolled: w.perRegion == 1, EnrollAt: w.now, DeleteOnTermination: true}
			nb.forget()
			w.boxes = append(w.boxes, nb)
			w.logf("box %d replaces the destroyed box in region %d", nb.ID, r)
		}
	}
}