package hostlife

import (
	"fmt"
	"strings"
)

// ---------- the built completion protocol (L18 joint round, findings 2 to 4) ----------
//
// Finding 2: a relocation does not end with the probe still running. The
// built sequence (1b-next 596fa8b1; hostrelocate Participant and Worker,
// host relocationStopProbe and relocationReleaseHold, hostclaim
// CompleteRelocation, completionBarrier) is:
//
//  1. probe stop (ruling 49): with the probe journaled (the model's
//     probed), the target persists the owner of the fence it will install
//     (probe-fence.json), installs that owned activator fence on an
//     unfenced host or meets its own (ruling 75; any other owner's fence
//     refuses with nothing stopped), stops the session's whole tree with
//     root's receipt and journals the stop (RecordProbeStop). Ruling 67
//     bounds it: no stop ReleaseDeadline after the probe passed is stuck;
//  2. the worker's serving CAS (CompleteRelocation) requires that stop and
//     records completed_relocation on the serving op;
//  3. the completion pass on the box it completed on (relocationReleaseHold,
//     rulings 66, 76, 80): C10's discharge of the participation, then the
//     row's hold released, then the owned fence lifted, then the mark
//     (MarkCompletedDischarged) last; found from the claim record, never
//     from the row, so a failure at any step is retried; run whenever claim
//     mode is on, region HA on or off (ruling 82);
//  4. the first serving wake after it is a separate admission that hands
//     over from the probe's receipt (N6), never main's stop;
//  5. until the mark, every start from serving refuses (ruling 77), except
//     with the completion box on lost or cited as a failover's fenced
//     source.
//
// Finding 3 (ruling 85, 1b-next a092ddc2): the source's row hold is
// reconcile's, not the release's (ruling 28 fences admission at the
// release's end), so a source whose reconcile runs only after the
// completion catches up: serving on another box, a completed relocation
// that did not complete here, and a park-local this box owes for exactly
// that op hold the row until the park-local retires it.
//
// Finding 4: today's fallbacks for claim asks C1 and C11. A failed target
// cannot release in probe-failed (QueueRelocationRelease needs the
// source's releasing; no failed-target obligation), so it reaches stuck at
// the release deadline; and after P's opening a later cold probe attempt
// is refused before it is counted (root pins every relocating launch to
// install_stamp), which is probe-failed. world.claimC1 and claimC11 are
// the switches that flip them when the asks land.

// probeOwner is the owner a probe stop persists and installs its fence
// under: the operation and the probe's session.
func probeOwner(op string, session int) string { return fmt.Sprintf("probe:%s/%d", op, session) }

// probeKey is the stop evidence key of op's probe stop: root's completed
// writer-session receipt in the box's guard journal (durable).
func probeKey(op string) string { return "probe:" + op }

// relocStopProbe is relocationStopProbe and RecordProbeStop (ruling 49,
// ruling 75): op.exec stops its passed probe's session while the op is
// still relocating. The fence's owner is persisted first; the owned fence
// is installed only on an unfenced host or met again (rule
// reloc-probe-fence-owner; the mutation installs over any fence); then
// the whole tree is stopped and the session's database closed cleanly, and
// with root's receipt the stop is journaled.
func (w *world) relocStopProbe(b *box, c *claim) {
	op, h := c.Op, c.Holder
	if h == nil || h.WriterBox != b.ID || op.Exec != b.ID {
		return
	}
	// ruling 87 (1b-next 6a4eadde, EndProbe): the release's own steps first
	// (flushStopped): the writers stopped, the database and the files
	// flushed, the positions recorded; only then the activator's fence
	// and root's stop. A restart before the record leaves no replicator,
	// so the flush refuses until ruling 67's deadline makes the op stuck,
	// as a release does. The mutation of reloc-probe-stop-flush stops at
	// once with nothing flushed (joint round 2's model)
	if !w.evidence(b, "probe-writers:"+op.ID) {
		w.stats["probe stop: a process survives"]++
		return
	}
	if on("reloc-probe-stop-flush") {
		if r := b.probeRecord; r == nil || r.Op != op.ID || r.Session != h.Session {
			tx, seq, ok := w.flushProbe(b, c)
			if !ok {
				return
			}
			b.probeRecord = &probeRecordEntry{Op: op.ID, Session: h.Session, TX: tx, Seq: seq}
			w.stats["probe stop: flushed and recorded"]++
		}
	}
	owner := probeOwner(op.ID, h.Session)
	if strings.HasPrefix(b.probeFence, probeKey(op.ID)+"/") && b.probeFence != owner {
		w.stats["probe stop: the persisted fence owner names another session"]++
		return
	}
	b.probeFence = owner // probe-fence.json, persisted before anything is asked of the activator
	if b.actFence != "" && b.actFence != owner {
		if on("reloc-probe-fence-owner") {
			// ErrFenceHeld (ruling 75 as ruling 87 amends it): the writers
			// are stopped and flushed, root is not asked, and the other
			// owner's fence is left as it is
			w.stats["probe stop: another owner's fence, root not asked"]++
			w.logf("relocation %s: box %d's probe stop meets %s's fence; flushed, root not asked", op.ID, b.ID, b.actFence)
			return
		}
		w.stats["probe stop: another owner's fence replaced (mutation)"]++
	}
	b.actFence = owner
	// root's writer-session stop and its receipt
	if !w.evidence(b, probeKey(op.ID)) {
		w.stats["probe stop: a process survives"]++
		return
	}
	if w.cas(c, func(n *claim) bool {
		ok := sameOp(n.Op.ID, op.ID) && n.Op.Kind == "relocating" && n.Op.Phase == "probed" && n.Op.Exec == b.ID && n.Holder.Session == h.Session && n.Op.ProbeStop == nil
		if ok {
			n.Op.ProbeStop = &probeStopEntry{Box: b.ID, Session: h.Session}
		}
		return ok
	}) {
		w.stats["probe stop journaled"]++
		w.logf("relocation %s: box %d stopped its probe session %d with root's receipt", op.ID, b.ID, h.Session)
	}
}

// relocComplete is the worker's CompleteRelocation: the serving CAS
// requires the journaled stop of exactly the probe's session (rule
// reloc-probe-stop; the mutation completes over the running probe, which
// strands its session across the kind change: INV-31).
func (w *world) relocComplete(c *claim) {
	op := c.Op
	if on("reloc-probe-stop") && (op.ProbeStop == nil || op.ProbeStop.Box != op.Exec || op.ProbeStop.Session != c.Holder.Session) {
		return
	}
	w.release(c, op.Exec)
}

// completionPass is relocationReleaseHold for every serving claim whose
// completion record names this box and is not marked (ruling 80). Each
// step is idempotent; a failure or a crash leaves the record unmarked and
// the next pass resumes. Rules: reloc-completion-claim-mode (ruling 82),
// reloc-completion-from-claim (ruling 80's finding), reloc-completion-
// discharge-first (ruling 66), reloc-completion-mark-last (ruling 80) and
// reloc-probe-fence-owner (the owned lift).
func (w *world) completionPass(b *box, c *claim) {
	d := c.Op.Completed
	if c.Op.Kind != "serving" || d == nil || d.Discharged || d.Box != b.ID || c.Holder == nil || c.Op.Exec != b.ID || c.Holder.WriterBox != b.ID {
		return
	}
	if w.haOff && !on("reloc-completion-claim-mode") {
		return // the mutation: the pass runs only with region HA on
	}
	if !on("reloc-completion-from-claim") && (b.Row == nil || b.Row.HoldOp != d.Op) {
		return // the mutation: the completion is found from the row's hold, which a finished release no longer shows
	}
	if b.Row == nil {
		w.stats["completion: no row, refused"]++
		return
	}
	if b.Row.HoldOp != "" && b.Row.HoldOp != d.Op {
		w.stats["completion: the row is held by another op, refused"]++
		return
	}
	owner := probeOwner(d.Op, d.Session)
	if b.probeFence != owner {
		w.stats["completion: no probe-fence owner, refused"]++
		return
	}
	held := b.Row.HoldOp == d.Op
	if !on("reloc-completion-mark-last") && !w.markCompleted(b, c) {
		return // the mutation: the mark first
	}
	releaseHold := func() {
		if b.Row != nil && b.Row.HoldOp == d.Op {
			b.Row.Status, b.Row.HoldOp = "ready", ""
			w.logf("box %d: completion of %s released the row's hold", b.ID, d.Op)
		}
	}
	lift := func() {
		if b.actFence == owner || !on("reloc-probe-fence-owner") {
			b.actFence = "" // LiftFenceAs: only the owner's fence (the mutation lifts any)
		}
	}
	if !on("reloc-completion-discharge-first") {
		// the mutation: the hold released and the fence lifted before the
		// discharge, so a serving admission can move the holder first
		releaseHold()
		lift()
		if !w.dischargeCompleted(b, c, false) {
			return
		}
	} else {
		if held && c.Holder.Session != d.Session {
			w.stats["completion: the holder is not the stopped probe session, refused"]++
			return
		}
		if !w.dischargeCompleted(b, c, held) {
			return
		}
		if w.crashed(b, "discharge") {
			return
		}
		releaseHold()
		if w.crashed(b, "release") {
			return
		}
		lift()
		if w.crashed(b, "lift") {
			return
		}
	}
	if on("reloc-completion-mark-last") {
		w.markCompleted(b, c)
	}
}

// crashed is a witness's agent crash after the named step of the
// completion pass (a restart: the activator's fences are gone).
func (w *world) crashed(b *box, step string) bool {
	if b.crashAfter != step {
		return false
	}
	b.crashAfter = ""
	b.forget()
	w.logf("witness: box %d's agent crashed after the completion's %s", b.ID, step)
	return true
}

// dischargeCompleted is C10's discharge (completedRelocationDischarge,
// DischargeCompleted): keyed by the persisted binding, it needs, while the
// row is still held for the op, root's receipt for the stopped probe
// session; it closes main's participation slot and clears the .hostexec
// record (the model's execRecord). A replay after the hold's release finds
// no slot, which is positive evidence. dischargeFails is a witness's
// failing discharge.
func (w *world) dischargeCompleted(b *box, c *claim, held bool) bool {
	d := c.Op.Completed
	if b.dischargeFails {
		w.stats["completion: discharge failed"]++
		return false
	}
	if held && !b.StopEvidence[probeKey(d.Op)] {
		w.stats["completion: no receipt for the probe's stop, discharge refused"]++
		return false
	}
	if b.execRecord == d.Op {
		b.execRecord, b.cleanupPending = "", false
	}
	if b.owedDischarge == d.Op {
		b.owedDischarge = ""
		w.stats["completion: participation discharged"]++
	}
	return true
}

// markCompleted is MarkCompletedDischarged: the serving op's record is
// marked, which lets any successor start. markFails is a witness's failed
// mark (the CAS lost, or the store refused it).
func (w *world) markCompleted(b *box, c *claim) bool {
	d := c.Op.Completed
	if b.markFails {
		b.markFails = false
		w.stats["completion: mark failed"]++
		return false
	}
	if w.cas(c, func(n *claim) bool {
		x := n.Op.Completed
		ok := n.Op.Kind == "serving" && x != nil && x.Op == d.Op && !x.Discharged
		if ok {
			cp := *x
			cp.Discharged = true
			n.Op.Completed = &cp
		}
		return ok
	}) {
		w.stats["completion marked discharged"]++
		w.logf("box %d: completion of %s marked discharged", b.ID, d.Op)
		return true
	}
	return false
}

// completionBarrier is completionBarrier (ruling 77; rule
// reloc-completion-barrier): true refuses a start from serving while the
// serving op's completed relocation is not marked, unless its box is on
// lost or is the source the start cites as fenced (fenced, -1 none).
func (w *world) completionBarrier(c *claim, fenced int) bool {
	d := c.Op.Completed
	if c.Op.Kind != "serving" || d == nil || d.Discharged || !on("reloc-completion-barrier") {
		return false
	}
	if contains(c.Lost, d.Box) || fenced >= 0 && fenced == d.Box {
		return false
	}
	w.stats["start refused: a completed relocation is not discharged"]++
	return true
}

// completedHandover is N6 (wireCompletedHandover): the first serving wake
// on the box a relocation completed on, while the holder is still the
// stopped probe session, hands over from the probe's receipt and stops
// nothing. It reports whether the wake may go on. The mutation of
// reloc-completed-handover routes it to main's stop, which installs a
// serving-shaped fence over the probe's and re-stops S, whose stop root
// already completed under the probe's fence: the handover refuses for
// good, and the fence is never the probe's to lift.
func (w *world) completedHandover(b *box, c *claim) bool {
	d := c.Op.Completed
	if c.Op.Kind != "serving" || d == nil || d.Box != b.ID || c.Holder == nil || c.Holder.Session != d.Session || c.Holder.WriterBox != b.ID {
		return true // not the completed prior: main's path
	}
	if !on("reloc-completed-handover") {
		if b.actFence == "" || strings.HasPrefix(b.actFence, "probe:") {
			b.actFence = fmt.Sprintf("main:%d", d.Session)
		}
		w.stats["serving wake: main's stop of the completed prior refused (mutation)"]++
		return false
	}
	if !b.StopEvidence[probeKey(d.Op)] {
		w.stats["serving wake: no receipt for the completed probe, refused"]++
		return false // absence is never evidence
	}
	w.stats["serving wake: handed over from the probe's receipt"]++
	return true
}

// completionHolds: the row's hold is the unmarked completion's own, which
// only the completion pass releases, after the discharge (ruling 66); the
// mutation of reloc-completion-discharge-first lets reconcile release it
// as a stale hold.
func (w *world) completionHolds(c *claim, b *box) bool {
	d := c.Op.Completed
	return d != nil && !d.Discharged && d.Box == b.ID && b.Row != nil && b.Row.HoldOp == d.Op && on("reloc-completion-discharge-first")
}

// catchUpHold is reconcile's completedHold (finding 3, ruling 85): with
// region HA on, a box that is not the serving executor, not on lost, and
// owes a park-local for exactly the serving op's completed relocation
// (which completed elsewhere) holds its row for that op until the
// park-local retires it. It reports whether the case applied. The
// mutation drops the hold, and the row a delayed source's release left
// ready stays ready (INV-33: the edge routes requests to a box that does
// not serve).
func (w *world) catchUpHold(b *box, c *claim) bool {
	d := c.Op.Completed
	if w.haOff || c.Op.Kind != "serving" || d == nil || d.Box == b.ID || c.Op.Exec == b.ID || contains(c.Lost, b.ID) || b.Row == nil {
		return false
	}
	owes := false
	for _, o := range c.Obl {
		if o.Kind == "park-local" && o.Box == b.ID && o.Op == d.Op {
			owes = true
		}
	}
	if !owes {
		return false
	}
	if b.Row.Status == "ready" && on("reloc-completed-catchup-hold") {
		b.Row.Status, b.Row.HoldOp = "relocating", d.Op
		w.stats["catch-up hold: a delayed source held its row"]++
		w.logf("box %d: reconcile caught up: its row is held for completed relocation %s until its park-local", b.ID, d.Op)
	}
	if b.Row.Status == "ready" {
		w.violate("INV-33: box %d's reconcile left its row ready after relocation %s completed on box %d, while it owes that op's park-local (the edge routes requests to a box that does not serve)", b.ID, d.Op, d.Box)
	}
	return true
}

// checkCompletion is INV-31 (rulings 49, 66, 77): while a box has not
// discharged its participation in a completed relocation, nothing of the
// host runs on it (the stopped probe session stays the holder until C10
// discharges it, and a session the kind change stranded is never
// discharged), and no operation names it as a participant (main's Prepare
// answers ErrLost). A box on lost, or destroyed, holds nothing a
// successor needs.
func (w *world) checkCompletion() {
	c := w.claim
	if c != nil && c.Holder != nil && c.Op.Kind == "serving" {
		if d := c.Op.Completed; d != nil && !d.Discharged && c.Holder.Session != d.Session {
			if b := w.boxByID(d.Box); b != nil && b.owedDischarge == d.Op && !b.Destroyed && !contains(c.Lost, b.ID) {
				w.violate("INV-31: a writer was admitted (session %d, box %d) before box %d discharged its participation in completed relocation %s, whose holder must still be the stopped probe session %d (ruling 66)", c.Holder.Session, c.Holder.WriterBox, b.ID, d.Op, d.Session)
				return
			}
		}
	}
	for _, b := range w.boxes {
		if b.owedDischarge == "" || b.Destroyed || c != nil && contains(c.Lost, b.ID) {
			continue
		}
		if len(b.Procs) > 0 {
			w.violate("INV-31: the host runs on box %d before its participation in completed relocation %s was discharged (rulings 49, 66)", b.ID, b.owedDischarge)
			return
		}
		if c != nil && c.Op.Kind != "serving" && c.Op.Kind != "" && opNames(c.Op, b.ID) {
			w.violate("INV-31: %s %s names box %d before its participation in completed relocation %s was discharged (ruling 77: main's Prepare answers ErrLost)", c.Op.Kind, c.Op.ID, b.ID, b.owedDischarge)
			return
		}
	}
}

// opNames: op names box id as its executor or a participant, or as its
// source or target in the kinds that set them.
func opNames(op opState, id int) bool {
	if op.Exec == id || contains(op.Parts, id) {
		return true
	}
	switch op.Kind {
	case "moving", "landing", "restoring", "reverting", "pushing", "relocating":
		return op.Src == id || op.Tgt == id
	}
	return false
}

// c1Fallback: claim ask C1 is open, so a failed target has no release
// obligation and its release in probe-failed is refused (rule
// reloc-c1-failed-release-refused; the mutation releases as if C1 had
// landed).
func (w *world) c1Fallback() bool { return !w.claimC1 && on("reloc-c1-failed-release-refused") }

// c11Fallback: claim ask C11 is open, so root launches no cold probe
// attempt after P's opening (rule reloc-c11-later-attempt-refused; the
// mutation counts and makes it as if C11 had landed).
func (w *world) c11Fallback() bool { return !w.claimC11 && on("reloc-c11-later-attempt-refused") }

// verifiedCompletionFence is ProductionAuthority.completionFence (ruling
// 86, 1b-next ca282378): a delete or a purge passes StartGuard the
// completion box whose termination fence it verified, as a failover
// passes its fenced source, or -1. Verified means the box is in the
// roster (enrolled), its box-fences record exists (a regional worker
// observed the termination: FenceObservedAt) and the cloud says it is
// terminated (VerifyFenceForBox); never inferred from a missing fence.
// The mutation of reloc-completion-fenced-exempt verifies nothing, as
// before ruling 86: a completion box terminated before its discharge
// while a purge holds admission closed blocks the purge, and the purge's
// closed admission blocks the failover (seed 2224 / 3,000, two boxes).
func (w *world) verifiedCompletionFence(c *claim) int {
	d := c.Op.Completed
	if c.Op.Kind != "serving" || d == nil || d.Discharged || contains(c.Lost, d.Box) || !on("reloc-completion-fenced-exempt") {
		return -1
	}
	x := w.boxByID(d.Box)
	if x == nil || !x.Enrolled || !x.Terminated || x.FenceObservedAt == 0 {
		return -1
	}
	w.stats["start: the completion box's termination fence verified"]++
	return x.ID
}

// completedPrior is relocationCompletedHandover.completedPrior: the claim
// serves on b with a completion recorded on b, and the holder is still the
// stopped probe session.
func (w *world) completedPrior(b *box, c *claim) bool {
	d := c.Op.Completed
	return c.Op.Kind == "serving" && d != nil && d.Box == b.ID && c.Holder != nil && c.Holder.Session == d.Session && c.Holder.WriterBox == b.ID
}

// closeLaunches is StopWriters' first step: the current attempt's and
// session's launch journal is closed, so the stopped session is never
// relaunched (an admitted start still in progress is abandoned). It is
// not root's admission fence, which Collect writes later (relFence).
func (w *world) closeLaunches(b *box, op string, session int) {
	if b.launchClosedOp != op || b.launchClosedSession != session {
		b.launchClosedOp, b.launchClosedSession = op, session
		w.logf("box %d: relocation %s closed session %d's launch journal", b.ID, op, session)
	}
	if b.open != nil && b.open.Session == session {
		b.open = nil
		w.stats["release: a start in progress abandoned with its session"]++
	}
}

// probeRecordEntry is the probe stop's recorded positions (ruling 87,
// ReleasePositions under the probe's own stop request): durable, so a
// restart after the record stops without flushing again.
type probeRecordEntry struct {
	Op          string
	Session, TX int
	Seq         int
}

// flushProbe is flushStopped's database and files steps for the probe
// session (ruling 87, the release's FlushDB and FlushFiles): they need
// the session's own replicator and the box's own files directory; with
// either gone (a restart before the record) the flush refuses. The
// pending writes ship and the listing must reach the stopped database's
// position; the files and wake state are published as one manifest of the
// generation, acknowledged. It reports the positions once both landed.
func (w *world) flushProbe(b *box, c *claim) (tx, seq int, ok bool) {
	h := c.Holder
	db := b.DB
	if db == nil || db.Inc != h.Inc || db.Gen != h.Gen || db.Epoch != b.Epoch || db.Session != h.Session || b.Files == nil || b.Files.Inc != h.Inc {
		w.stats["probe stop: no replicator or files directory and no record, refused"]++
		return 0, 0, false
	}
	if len(db.Pending) > 0 {
		bl := w.deltaOf(db, db.Inc)
		db.FirstFull = false
		db.Queue = append(db.Queue, bl)
		db.Pos, db.Pending = bl.Max, nil
	}
	w.ship(b)
	if len(db.Queue) > 0 || db.InFlight != nil || w.remoteMax(w.buckets[b.Region], dataKey(h.Location, h.Gen, "")) < db.Pos {
		w.stats["probe stop: waits for the flush"]++
		return 0, 0, false
	}
	files := b.Files.clone()
	if b.pubGen != h.Gen {
		b.pubGen, b.pubSeq, b.releaseSeq = h.Gen, 0, 0
	}
	if !w.ensureObject(b, h, files) {
		w.stats["probe stop: waits for its files object"]++
		return 0, 0, false
	}
	if b.releaseSeq == 0 {
		b.pubSeq++
		b.releaseSeq = b.pubSeq
	}
	w.issued++
	bl := &blob{Kind: "files", Inc: h.Inc, Gen: h.Gen, Seq: b.releaseSeq, Files: files.clone(), Timers: append([]int(nil), b.Timers...), Base: w.issued}
	if !w.ackedPut(b, manifestKey(h.Location, h.Gen, b.releaseSeq), bl, true) {
		w.stats["probe stop: waits for its manifest"]++
		return 0, 0, false
	}
	seq, b.releaseSeq = b.releaseSeq, 0
	b.published = filesState(b.Timers, files)
	return db.Pos, seq, true
}