// Mgmt
// Copyright (C) James Shubin and the project contributors
// Written by James Shubin <james@shubin.ca> and the project contributors
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//
// Additional permission under GNU GPL version 3 section 7
//
// If you modify this program, or any covered work, by linking or combining it
// with embedded mcl code and modules (and that the embedded mcl code and
// modules which link with this program, contain a copy of their source code in
// the authoritative form) containing parts covered by the terms of any other
// license, the licensors of this program grant you additional permission to
// convey the resulting work. Furthermore, the licensors of this program grant
// the original author, James Shubin, additional permission to update this
// additional permission if he deems it necessary to achieve the goals of this
// additional permission.

package graph

import (
	"context"
	"fmt"
	"strings"
	"sync"
	"time"

	"github.com/purpleidea/mgmt/engine"
	engineUtil "github.com/purpleidea/mgmt/engine/util"
	"github.com/purpleidea/mgmt/pgraph"
	"github.com/purpleidea/mgmt/util/errwrap"

	"golang.org/x/time/rate"
)

// OKTimestamp returns true if this vertex can run right now.
func (obj *Engine) OKTimestamp(vertex pgraph.Vertex) bool {
	return len(obj.BadTimestamps(vertex)) == 0
}

// BadTimestamps returns the list of vertices that are causing our timestamp to
// be bad.
func (obj *Engine) BadTimestamps(vertex pgraph.Vertex) []pgraph.Vertex {
	obj.tlock.RLock()
	state := obj.state[vertex]
	obj.tlock.RUnlock()

	vs := []pgraph.Vertex{}
	state.mutex.RLock()   // concurrent read start
	ts := state.timestamp // race
	state.mutex.RUnlock() // concurrent read end
	// these are all the vertices pointing TO vertex, eg: ??? -> vertex
	for _, v := range obj.graph.IncomingGraphVertices(vertex) {
		obj.tlock.RLock()
		state := obj.state[v]
		obj.tlock.RUnlock()

		// If the vertex has a greater timestamp than any prerequisite,
		// then we can't run right now. If they're equal (eg: initially
		// with a value of 0) then we also can't run because we should
		// let our pre-requisites go first.
		state.mutex.RLock()   // concurrent read start
		t := state.timestamp  // race
		state.mutex.RUnlock() // concurrent read end
		if obj.Debug {
			obj.Logf("OKTimestamp: %d >= %d (%s): !%t", ts, t, v.String(), ts >= t)
		}
		if ts >= t {
			//return false
			vs = append(vs, v)
		}
	}
	return vs // formerly "true" if empty
}

// errWatchRestart is the sentinel which the Watch retry loop hands to the
// process loop when Watch failed and is about to restart. Receiving it (which
// can only happen when no Process is running) withdraws the started state until
// the replacement Watch sends its initial startup event.
var errWatchRestart = engine.Error("watch restart")

// Process is the primary function to execute a particular vertex in the graph.
func (obj *Engine) Process(ctx context.Context, vertex pgraph.Vertex) error {
	res, isRes := vertex.(engine.Res)
	if !isRes {
		return fmt.Errorf("vertex is not a Res")
	}

	// The user asked us to stop waiting, so don't start any new work. Watch
	// keeps running until we're paused, and every event it sends would
	// otherwise start a fresh CheckApply that we'd then have to wait for.
	// We leave the state dirty, since we didn't converge it.
	if obj.interrupt.Load() {
		if obj.Debug {
			obj.Logf("%s: interrupted, skipping Process", vertex)
		}
		return ctx.Err()
	}

	obj.tlock.RLock()
	state := obj.state[vertex]
	obj.tlock.RUnlock()

	// backpoke! (can be async)
	if vs := obj.BadTimestamps(vertex); len(vs) > 0 {
		// back poke in parallel (sync b/c of waitgroup)
		wg := &sync.WaitGroup{}
		for _, v := range obj.graph.IncomingGraphVertices(vertex) {
			if !pgraph.VertexContains(v, vs) { // only poke what's needed
				continue
			}

			// doesn't really need to be in parallel, but we can...
			wg.Add(1)
			go func(vv pgraph.Vertex) {
				defer wg.Done()
				obj.state[vv].Poke() // async
			}(v)

		}
		wg.Wait()

		// can't continue until timestamp is in sequence, defer for now
		return engine.ErrBackPoke
	}

	// semaphores!
	// These shouldn't ever block an exit, since the graph should eventually
	// converge causing their them to unlock. More interestingly, since they
	// run in a DAG alphabetically, there is no way to permanently deadlock,
	// assuming that resources individually don't ever block from finishing!
	// The exception is that semaphores with a zero count will always block!
	// TODO: Add a close mechanism to close/unblock zero count semaphores...
	semas := res.MetaParams().Sema
	if obj.Debug && len(semas) > 0 {
		obj.Logf("%s: Sema: P(%s)", res, strings.Join(semas, ", "))
	}
	if err := obj.semaLock(semas); err != nil { // lock
		// NOTE: in practice, this might not ever be truly necessary...
		return fmt.Errorf("shutdown of semaphores")
	}
	defer obj.semaUnlock(semas) // unlock
	if obj.Debug && len(semas) > 0 {
		defer obj.Logf("%s: Sema: V(%s)", res, strings.Join(semas, ", "))
	}

	// sendrecv!
	// connect any senders to receivers and detect if values changed
	// this actually checks and sends into resource trees recursively...

	// XXX: This code is duplicated in the fancier autogrouping code below!
	//if res, ok := vertex.(engine.RecvableRes); ok {
	//	if obj.Debug {
	//		obj.Logf("SendRecv: %s", res) // receiving here
	//	}
	//	if updated, err := SendRecv(res, nil); err != nil {
	//		return errwrap.Wrapf(err, "could not SendRecv")
	//	} else if len(updated) > 0 {
	//		//for _, s := range graph.UpdatedStrings(updated) {
	//		//	obj.Logf("SendRecv: %s", s)
	//		//}
	//		for r, m := range updated { // map[engine.RecvableRes]map[string]*engine.Send
	//			v, ok := r.(pgraph.Vertex)
	//			if !ok {
	//				continue
	//			}
	//			_, stateExists := obj.state[v] // autogrouped children probably don't have a state
	//			if !stateExists {
	//				continue
	//			}
	//			for s, send := range m {
	//				if !send.Changed {
	//					continue
	//				}
	//				obj.Logf("Send/Recv: %v.%s -> %v.%s", send.Res, send.Key, r, s)
	//				// if send.Changed == true, at least one was updated
	//				// invalidate cache, mark as dirty
	//				obj.state[v].setDirty()
	//				//break // we might have more vertices now
	//			}
	//
	//			// re-validate after we change any values
	//			if err := engine.Validate(r); err != nil {
	//				return errwrap.Wrapf(err, "failed Validate after SendRecv")
	//			}
	//		}
	//	}
	//}

	// Send/Recv *can* receive from someone that was grouped! The sender has
	// to use *their* send/recv handle/implementation, which has to be setup
	// properly by the parent resource during Init(). See: http:server:flag.
	// NOTE: SendRecv recurses into everything grouped in here, so we don't
	// collect those resources ourselves. It must run even when this vertex
	// can't receive on its own fields, because a groupable parent (such as
	// http:server) is commonly not recvable while its children are.
	if updated, err := SendRecv(res, nil); err != nil {
		return errwrap.Wrapf(err, "could not SendRecv")
	} else if len(updated) > 0 {
		//for _, s := range graph.UpdatedStrings(updated) {
		//	obj.Logf("SendRecv: %s", s)
		//}
		for r, m := range updated { // map[engine.RecvableRes]map[string]*engine.Send
			v, ok := r.(pgraph.Vertex)
			if !ok {
				continue
			}
			state, stateExists := obj.state[v] // autogrouped children probably don't have a state
			if !stateExists {
				// An autogrouped child doesn't have a state of
				// its own, so the vertex which owns it is the
				// one to dirty. That parent CheckApply is what
				// runs the child, so if we skipped this, then
				// it could stay clean and the received value
				// would never get applied.
				state = obj.state[vertex]
			}
			for s, send := range m {
				if !send.Changed {
					continue
				}
				obj.Logf("Send/Recv: %v.%s -> %v.%s", send.Res, send.Key, r, s)
				// if send.Changed == true, at least one was updated
				// invalidate cache, mark as dirty
				state.setDirty()
				//break // we might have more vertices now
			}
			if !stateExists {
				continue // only re-validate our own
			}

			// re-validate after we change any values
			if err := engine.Validate(r); err != nil {
				return errwrap.Wrapf(err, "failed Validate after SendRecv")
			}
		}
	}
	// XXX: this might not work with two merged "CompatibleRes" resources...
	// XXX: fix that so we can have the mappings to do it in lang/interpret.go ?

	var ok = true
	var applied = false              // did we run an apply?
	var noop = res.MetaParams().Noop // lookup the noop value
	var refresh bool
	var checkOK bool
	var err error

	// lookup the refresh (notification) variable
	refresh = obj.RefreshPending(vertex) // do i need to perform a refresh?
	refreshableRes, isRefreshableRes := vertex.(engine.RefreshableRes)
	if isRefreshableRes {
		refreshableRes.SetRefresh(refresh) // tell the resource
	}

	// Run the exported resource exporter!
	var exportOK bool
	var exportErr error
	wg := &sync.WaitGroup{}
	wg.Add(1)
	// (Run this concurrently with the CheckApply related stuff below...)
	go func() {
		defer wg.Done()
		// doesn't really need to be in parallel, but we can...
		exportOK, exportErr = obj.Exporter.Export(ctx, res)
	}()

	// Did we actually run CheckApply? If we didn't, then the resource never
	// got the chance to look at its send/recv changed flags, and so we must
	// leave them alone for a subsequent run to consume.
	ranCheckApply := false

	// Check cached state, to skip CheckApply, but can't skip if refreshing!
	// If the resource doesn't implement refresh, skip the refresh test.
	// FIXME: if desired, check that we pass through refresh notifications!
	if (!refresh || !isRefreshableRes) && state.isStateOK.Load() { // mutex RLock/RUnlock
		checkOK, err = true, nil

	} else if noop && (refresh && isRefreshableRes) { // had a refresh to do w/ noop!
		checkOK, err = false, nil // therefore the state is wrong

	} else if res.MetaParams().Hidden {
		// We're not running CheckApply
		if obj.Debug {
			obj.Logf("%s: Hidden", res)
		}
		checkOK, err = true, nil // default

	} else {
		// run the CheckApply!
		if obj.Debug {
			obj.Logf("%s: CheckApply(%t)", res, !noop)
		}
		// if this fails, don't UpdateTimestamp()
		ranCheckApply = true
		checkOK, err = safeCheckApply(ctx, res, !noop)
		if !checkOK && obj.Debug { // don't log on (checkOK == true)
			obj.Logf("%s: CheckApply(%t): Return(%t, %s)", res, !noop, checkOK, engineUtil.CleanError(err))
		}
	}
	wg.Wait()

	// The resource has now had its chance to consume any send/recv changed
	// flags, so turn them off. We wait for the exporter above to finish
	// first, because it reads those flags too. If CheckApply errored, then
	// we leave them on, because the Worker retry loop runs us again, and
	// SendRecv won't report the same transfer a second time. This is also
	// why we don't do this when CheckApply was skipped entirely.
	if ranCheckApply && err == nil {
		ClearRecv(res)
	}

	checkOK = checkOK && exportOK // always combine
	if err == nil {               // If CheckApply didn't error, look at exportOK.
		// This is because if CheckApply errors we don't need to care or
		// tell anyone about an exporting error.
		err = exportErr
	}

	if checkOK && err != nil { // should never return this way
		return fmt.Errorf("%s: resource programming error: CheckApply(%t): %t, %+v", res, !noop, checkOK, err)
	}

	if !checkOK { // something changed, restart timer
		state.cuid.ResetTimer() // activity!
		if obj.Debug {
			obj.Logf("%s: converger: reset timer", res)
		}
	}

	// if CheckApply ran without noop and without error, state should be good
	if !noop && err == nil { // aka !noop || checkOK
		state.tuid.StartTimer()
		//state.mutex.Lock()
		state.isStateOK.Store(true) // reset
		//state.mutex.Unlock()
		if refresh {
			obj.SetUpstreamRefresh(vertex, false) // refresh happened, clear the request
			if isRefreshableRes {
				refreshableRes.SetRefresh(false)
			}
		}
	}

	if !checkOK { // if state *was* not ok, we had to have apply'ed
		if err != nil { // error during check or apply
			ok = false
		} else {
			applied = true
		}
	}

	// when noop is true we always want to update timestamp
	if noop && err == nil {
		ok = true
	}

	if ok {
		// did we actually do work?
		activity := applied
		if noop {
			activity = false // no we didn't do work...
		}

		if activity { // add refresh flag to downstream edges...
			obj.SetDownstreamRefresh(vertex, true)
		}

		// poke! (should (must?) be sync)
		wg := &sync.WaitGroup{}
		// update this timestamp *before* we poke or the poked
		// nodes might fail due to having a too old timestamp!
		state.mutex.Lock()                      // concurrent write start
		state.timestamp = time.Now().UnixNano() // update timestamp (race)
		state.mutex.Unlock()                    // concurrent write end
		for _, v := range obj.graph.OutgoingGraphVertices(vertex) {
			if !obj.OKTimestamp(v) {
				// there is at least another one that will poke this...
				continue
			}

			// If we're pausing (or exiting) then we can skip poking
			// so that the graph doesn't go on running forever until
			// it's completely done. This is an optional feature and
			// we can select it via ^C on user exit or via the GAPI.
			if obj.fastPause.Load() {
				obj.Logf("%s: fast pausing, poke skipped", res)
				continue
			}

			// poke each vertex individually, in parallel...
			wg.Add(1)
			go func(vv pgraph.Vertex) {
				defer wg.Done()
				obj.state[vv].Poke()
			}(v)
		}
		wg.Wait()
	}

	return err
}

// Worker is the common run frontend of the vertex. It handles all of the retry
// and retry delay common code, and ultimately returns the final status of this
// vertex execution. This function cannot be "re-run" for the same vertex. The
// retry mechanism stuff happens inside of this. To actually "re-run" you need
// to remove the vertex and build a new one. The engine guarantees that we do
// not allow CheckApply to run while we are paused. That is enforced here.
func (obj *Engine) Worker(vertex pgraph.Vertex) error {
	res, isRes := vertex.(engine.Res)
	if !isRes {
		return fmt.Errorf("vertex is not a resource")
	}

	obj.tlock.RLock()
	state := obj.state[vertex]
	obj.tlock.RUnlock()

	// bonus safety check
	if res.MetaParams().Burst == 0 && !(res.MetaParams().Limit == rate.Inf) { // blocked
		return fmt.Errorf("permanently limited (rate != Inf, burst = 0)")
	}

	// initialize or reinitialize the meta state for this resource uid
	// if we're using a Hidden resource, we don't support this feature
	// TODO: should we consider supporting it? is it really necessary?
	// XXX: to support this for Hidden, we'd need to handle dupe names
	metas := &engine.MetaState{
		CheckApplyRetry: res.MetaParams().Retry, // lookup the retry value
	}
	if !res.MetaParams().Hidden {
		// Skip this if Hidden since we can have a hidden res that has
		// the same kind+name as a regular res, and this would conflict.
		obj.mlock.Lock()
		if _, exists := obj.metas[engine.PtrUID(res)]; !exists || res.MetaParams().Reset {
			obj.metas[engine.PtrUID(res)] = &engine.MetaState{
				CheckApplyRetry: res.MetaParams().Retry, // lookup the retry value
			}
		}
		metas = obj.metas[engine.PtrUID(res)] // handle
		obj.mlock.Unlock()
	}

	//defer close(state.stopped) // done signal

	state.cuid = obj.Converger.Register()
	state.tuid = obj.Converger.Register()
	// must wait for all users of the cuid to finish *before* we unregister!
	// as a result, this defer happens *before* the below wait group Wait...
	defer state.cuid.Unregister()
	defer state.tuid.Unregister()

	defer state.wg.Wait() // this Worker is the last to exit!

	state.wg.Add(1)
	go func() {
		defer state.wg.Done()
		defer close(state.eventsChan) // we close this on behalf of res

		// This is a close reverse-multiplexer. If any of the channels
		// close, then it will cause the doneCtx to cancel. That way,
		// multiple different folks can send a close signal, without
		// every worrying about duplicate channel close panics.
		state.wg.Add(1)
		go func() {
			defer state.wg.Done()

			// reverse-multiplexer: any close, causes *the* close!
			select {
			case <-state.processDone:
			case <-state.watchDone:
			case <-state.limitDone:
			case <-state.retryDone:
			case <-state.removeDone:
			case <-state.eventsDone:
			}

			// the main "done" signal gets activated here!
			state.doneCtxCancel() // cancels doneCtx
		}()

		var err error
		var retry = res.MetaParams().Retry // lookup the retry value
		var delay uint64
		for { // retry loop
			// a retry-delay was requested, wait, but don't block events!
			if delay > 0 {
				errDelayExpired := engine.Error("delay exit")
				err = func() error { // slim watch main loop
					//nolint:gosec // G115: delay is trusted operator config in ms; only wraps above 2^63
					timer := time.NewTimer(time.Duration(delay) * time.Millisecond)
					defer state.init.Logf("the Watch delay expired!")
					defer timer.Stop() // it's nice to cleanup
					for {
						select {
						case <-timer.C: // the wait is over
							return errDelayExpired // special

						case <-state.doneCtx.Done():
							return nil
						}
					}
				}()
				if err == errDelayExpired {
					delay = 0 // reset
					continue
				}

			} else if res.MetaParams().Hidden {
				// We're not running Watch
				if obj.Debug {
					obj.Logf("%s: Hidden", res)
				}
				state.cuid.StartTimer() // TODO: Should we do this?
				err = state.hidden(state.doneCtx)
				state.cuid.StopTimer() // TODO: Should we do this?

			} else if interval := res.MetaParams().Poll; interval > 0 { // poll instead of watching :(
				state.cuid.StartTimer()
				err = state.poll(state.doneCtx, interval)
				state.cuid.StopTimer() // clean up nicely

			} else if interval := res.MetaParams().Poll; interval < 0 { // run once instead of watching
				state.cuid.StartTimer()
				err = state.once(state.doneCtx)
				state.cuid.StopTimer() // clean up nicely

			} else {
				state.cuid.StartTimer()
				if obj.Debug {
					obj.Logf("%s: Watch...", vertex)
				}
				err = safeWatch(state.doneCtx, res)  // run the watch normally
				err = errwrap.NoContextCanceled(err) // strip
				if s := engineUtil.CleanError(err); err != nil {
					obj.Logf("%s: Watch Error: %s", vertex, s)
				} else if obj.Debug {
					obj.Logf("%s: Watch Exited...", vertex)
				}
				state.cuid.StopTimer() // clean up nicely
			}
			if err == nil { // || err == engine.ErrClosed
				return // exited cleanly, we're done
			}
			if err == context.Canceled {
				return // we shutdown nicely on request
			}
			// we've got an error...
			delay = res.MetaParams().Delay

			// A failed Watch must never overlap with a running
			// Process, and no new Process may run until our
			// replacement Watch sends its initial startup event.
			// Interrupt any running Process, and then synchronize
			// with the process loop through this sentinel, which it
			// can only consume when no Process is running.
			state.cancelProcess() // cancel any running Process
			if retry != 0 {       // we're going to retry Watch
				select {
				case state.eventsChan <- errWatchRestart:
				case <-state.doneCtx.Done():
					// we're shutting down instead
				}
			}

			if retry < 0 { // infinite retries
				continue
			}
			if retry > 0 { // don't decrement past 0
				retry--
				state.init.Logf("retrying Watch after %.4f seconds (%d left)", float64(delay)/1000, retry)
				continue
			}
			//if retry == 0 { // optional
			//	err = errwrap.Wrapf(err, "permanent watch error")
			//}
			break // break out of this and send the error
		} // for retry loop

		// this section sends an error...
		// If the CheckApply loop exits and THEN the Watch fails with an
		// error, then we'd be stuck here if exit signal didn't unblock!
		select {
		case state.eventsChan <- errwrap.Wrapf(err, "watch failed"):
			// send
		}
	}()

	// If this exits cleanly, we must unblock the reverse-multiplexer.
	// I think this additional close is unnecessary, but it's not harmful.
	defer close(state.eventsDone) // causes doneCtx to cancel
	limiter := rate.NewLimiter(res.MetaParams().Limit, res.MetaParams().Burst)
	var reserv *rate.Reservation
	var reterr error
	var failed bool  // has Process permanently failed?
	var closed bool  // has the resumeSignal channel closed?
	var started bool // has Watch sent the initial startup event?

	// Worker starts paused, so wait for resume signal before we CheckApply.
	// (it's okay to let Watch run immediately)
	select {
	case _, ok := <-state.resumeSignal: // channel closes
		if !ok {
			closed = true
		}

		// XXX: Should we have an easy exit here? Or let it exit below?
		//case <-state.doneCtx.Done():
		//	return nil
	}

Loop:
	for { // process loop
		// This is the main select where things happen and where we exit
		// from. It's similar to the two "satellite" select's which we
		// might spend some time in if we're retrying or rate limiting.
		// This select is also the main event receiver and is also the
		// only place where we read from the poke channel.
		select {
		case err, ok := <-state.eventsChan: // read from watch channel
			if !ok {
				return reterr // we only return when chan closes
			}
			if err == errWatchRestart {
				// Watch failed and is restarting. It already
				// interrupted any running Process, and since we
				// only receive between Process invocations, it
				// now knows that none is running. Require a new
				// initial event before we run any new Process.
				started = false
				continue
			}
			// If the Watch method exits with an error, then this
			// channel will get that error propagated to it, which
			// we then save so we can return it to the caller of us.
			if err != nil {
				failed = true
				close(state.watchDone)               // causes doneCtx to cancel
				reterr = errwrap.Append(reterr, err) // permanent failure
				continue
			}
			if obj.Debug {
				obj.Logf("event received")
			}
			// We mark the state dirty on receive, not on send,
			// since no Process can be running right now, so this
			// mark can't get clobbered by the completion of an
			// earlier Process.
			state.setDirty()
			started = true                           // the resource is now running
			reserv = limiter.ReserveN(time.Now(), 1) // one event
			// reserv.OK() seems to always be true here!

		case _, ok := <-state.pokeChan: // read from buffered poke channel
			if !ok { // we never close it
				panic("unexpected close of poke channel")
			}
			if obj.Debug {
				obj.Logf("poke received")
			}
			reserv = nil // we didn't receive a real event here...

		case _, ok := <-state.pauseSignal: // one message
			if !ok {
				state.pauseSignal = nil
				continue // this is not a new pause message
			}
			// NOTE: If we allowed a doneCtx below to let us out
			// of the resumeSignal wait, then we could loop around
			// and run this again, causing a panic. Instead of this
			// being made safe with a sync.Once, we instead run a
			// close() call inside of the vertexRemoveFn function,
			// which should unblock resumeSignal so we can shutdown.

			// we are paused now, and waiting for resume or exit...
			select {
			case _, ok := <-state.resumeSignal: // channel closes
				if !ok {
					closed = true
				}
				// resumed!
				// pass through to allow a Process to try to run
				// TODO: consider adding this fast pause here...
				//if obj.fastPause {
				//	obj.Logf("fast pausing on resume")
				//	continue
				//}
			}
		}

		// drop redundant pokes
		for len(state.pokeChan) > 0 {
			select {
			case <-state.pokeChan:
			default:
				// race, someone else read one!
			}
		}

		// don't Process anymore if we've already failed or shutdown...
		if failed || closed {
			continue Loop
		}

		// We must never Process (and CheckApply) before the initial
		// event from Watch tells us that the resource is started and
		// ready. A poke from a neighbouring resource can arrive before
		// that, and the resume signal passthrough can also get us here
		// early. It's safe to skip both, since the guaranteed initial
		// event always runs Process, which subsumes them.
		if !started {
			continue Loop
		}

		// limit delay
		d := time.Duration(0)
		if reserv != nil {
			d = reserv.DelayFrom(time.Now())
		}
		if reserv != nil && d > 0 { // delay
			state.init.Logf("limited (rate: %v/sec, burst: %d, next: %dms)", res.MetaParams().Limit, res.MetaParams().Burst, d/time.Millisecond)
			timer := time.NewTimer(d)
		LimitWait:
			for {
				// This "satellite" select doesn't need a poke
				// channel because we're already in "event
				// received" mode, and poke doesn't block.
				select {
				case <-timer.C: // the wait is over
					break LimitWait

				// consume other events while we're waiting...
				case e, ok := <-state.eventsChan: // read from watch channel
					if !ok {
						return reterr // we only return when chan closes
					}
					if e == errWatchRestart {
						started = false // wait for the new initial event
						continue
					}
					if e != nil {
						failed = true
						close(state.limitDone)             // causes doneCtx to cancel
						reterr = errwrap.Append(reterr, e) // permanent failure
						break LimitWait
					}
					if obj.Debug {
						obj.Logf("event received in limit")
					}
					state.setDirty() // on receive, see the main select
					// TODO: does this get added in properly?
					limiter.ReserveN(time.Now(), 1) // one event

				// this pause/resume block is the same as the upper main one
				case _, ok := <-state.pauseSignal:
					if !ok {
						state.pauseSignal = nil
						break LimitWait
					}
					select {
					case _, ok := <-state.resumeSignal: // channel closes
						if !ok {
							closed = true
						}
						// resumed!
					}
				}
			}
			timer.Stop() // it's nice to cleanup
			state.init.Logf("rate limiting expired!")
		}
		// don't Process anymore if we've already failed or shutdown...
		if failed || closed {
			continue Loop
		}
		if !started { // a restarting Watch withdrew our readiness...
			continue Loop
		}
		// end of limit delay

		// retry...
		var err error
		//var retry = res.MetaParams().Retry // lookup the retry value
		var delay uint64
	RetryLoop:
		for { // retry loop
			if delay > 0 {
				timer := time.NewTimer(time.Duration(delay) * time.Millisecond)
			RetryWait:
				for {
					// This "satellite" select doesn't need
					// a poke channel because we're already
					// in "event received" mode, and poke
					// doesn't block.
					select {
					case <-timer.C: // the wait is over
						break RetryWait

					// consume other events while we're waiting...
					case e, ok := <-state.eventsChan: // read from watch channel
						if !ok {
							return reterr // we only return when chan closes
						}
						if e == errWatchRestart {
							started = false // wait for the new initial event
							continue
						}
						if e != nil {
							failed = true
							close(state.retryDone)             // causes doneCtx to cancel
							reterr = errwrap.Append(reterr, e) // permanent failure
							break RetryWait
						}
						if obj.Debug {
							obj.Logf("event received in retry")
						}
						state.setDirty() // on receive, see the main select
						// TODO: does this get added in properly?
						limiter.ReserveN(time.Now(), 1) // one event

					// this pause/resume block is the same as the upper main one
					case _, ok := <-state.pauseSignal:
						if !ok {
							state.pauseSignal = nil
							break RetryWait
						}
						select {
						case _, ok := <-state.resumeSignal: // channel closes
							if !ok {
								closed = true
							}
							// resumed!
						}
					}
				}
				timer.Stop() // it's nice to cleanup
				delay = 0    // reset
				state.init.Logf("the CheckApply delay expired!")
			}

			// Without a retry delay the satellite select above
			// doesn't run at all, so a retry loop would spin
			// through Process without ever returning to a select
			// which reads the pause signal, and a Pause which
			// arrives while we retry would block until the retries
			// ran out. With Meta:retry set to -1 that is never.
			// Since Pause blocks on an unbuffered send, this
			// non-blocking receive catches one which is already
			// waiting for us, and costs nothing when there isn't
			// one. Between two Process runs is the only place
			// where a pause is allowed to land anyway, which is
			// exactly what the delay case above does too.
			select {
			case _, ok := <-state.pauseSignal:
				if !ok {
					// not a new pause
					state.pauseSignal = nil
					break // break out of our select
				}
				// we are paused now, and waiting for resume...
				select {
				case _, ok := <-state.resumeSignal: // channel closes
					if !ok {
						closed = true
					}
					// resumed!
				}

			default:
				// nobody is waiting on us, so carry on...
			}

			// don't Process anymore if we've already failed or shutdown...
			if failed || closed {
				continue Loop
			}
			if !started { // a restarting Watch withdrew our readiness...
				continue Loop
			}

			if obj.Debug {
				obj.Logf("Process(%s)", vertex)
			}
			backPoke := false
			// Run with a cancellable context, and register it, so
			// that a failing Watch can interrupt us to restart.
			processCtx, pCancel := context.WithCancel(state.doneCtx)
			state.registerProcessCancel(pCancel)
			err = obj.Process(processCtx, vertex)
			state.registerProcessCancel(nil)
			interrupted := processCtx.Err() != nil && state.doneCtx.Err() == nil
			pCancel() // cleanup the context
			if err == engine.ErrBackPoke {
				backPoke = true
				err = nil // for future code safety
			}
			if obj.Debug && backPoke {
				obj.Logf("Process(%s): BackPoke!", vertex)
			}
			if obj.Debug && !backPoke {
				obj.Logf("Process(%s): Return(%s)", vertex, engineUtil.CleanError(err))
			}
			if err == nil && !backPoke && res.MetaParams().RetryReset { // reset it on success!
				metas.CheckApplyRetry = res.MetaParams().Retry // lookup the retry value
			}
			if err == nil || backPoke {
				break RetryLoop
			}
			if interrupted {
				// Either a restarting Watch interrupted this
				// Process, or the user asked us to stop waiting
				// on it. Neither is a CheckApply failure, so it
				// must not count against the retry limit. In
				// the first case the initial event from the
				// replacement Watch re-runs it, and in the
				// second we're on our way out anyway.
				state.init.Logf("process interrupted")
				break RetryLoop
			}
			// we've got an error...
			delay = res.MetaParams().Delay

			if metas.CheckApplyRetry < 0 { // infinite retries
				continue
			}
			if metas.CheckApplyRetry > 0 { // don't decrement past 0
				metas.CheckApplyRetry--
				state.init.Logf(
					"retrying CheckApply after %.4f seconds (%d left)",
					float64(delay)/1000,
					metas.CheckApplyRetry,
				)
				continue
			}
			//if metas.CheckApplyRetry == 0 { // optional
			//	err = errwrap.Wrapf(err, "permanent process error")
			//}

			// It is important that we shutdown the Watch loop if
			// this dies. If Process fails permanently, we ask it
			// to exit right here... (It happens when we loop...)
			failed = true
			close(state.processDone)             // causes doneCtx to cancel
			reterr = errwrap.Append(reterr, err) // permanent failure
			continue

		} // retry loop

		// When this Process loop exits, it's because something has
		// caused Watch() to shutdown (even if it's our permanent
		// failure from Process), which caused this channel to close.
		// On or more exit signals are possible, and more than one can
		// happen simultaneously.

	} // process loop

	//return nil // unreachable
}

// safeCheckApply wraps a call to res.CheckApply with a panic recovery so that a
// buggy resource or upstream library doesn't bring down the entire engine.
func safeCheckApply(ctx context.Context, res engine.Res, apply bool) (checkOK bool, err error) {
	defer func() {
		if r := recover(); r != nil {
			checkOK = false
			err = fmt.Errorf("panic in CheckApply: %+v", r)
		}
	}()

	timeout := res.MetaParams().Timeout
	if timeout == 0 {
		return res.CheckApply(ctx, apply)
	}

	//nolint:gosec // G115: timeout is trusted operator config in ms; only wraps above 2^63
	duration := time.Duration(timeout) * time.Millisecond
	checkApplyCtx, cancel := context.WithTimeout(ctx, duration)
	defer cancel()

	// NOTE: We always pass through the same checkOK value below, because if
	// a resource made a mistake and returned (true, err) and we didn't pass
	// through the `true`, we'd be silently suppressing that resource bug...
	checkOK, err = res.CheckApply(checkApplyCtx, apply)
	if checkApplyCtx.Err() != context.DeadlineExceeded {
		return checkOK, err // return normally (err may even be nil)
	}
	// Timeout expired, resource returned, but we didn't return
	// DeadlineExceeded. This may be because the resource had a bug and
	// didn't properly propagate the error by returning ctx.Err() or it may
	// be because the last `<-ctx.Done()` check was earlier in the code and
	// the resource passed it. This is not a bug as long as workloads aren't
	// blocking for more than approximately 1,000ms.
	if err == nil {
		return checkOK, nil
	}
	if err == context.DeadlineExceeded { // It returned correctly! Nice...
		return checkOK, fmt.Errorf("timeout after %.3f seconds", duration.Seconds())
	}

	// This resource timed out, but returned an error. We're not sure
	// whether the timeout caused the error, or if it errored and happened
	// to timeout as well. Good resources return context.DeadlineExceeded,
	// when that is the root cause. Patch those if that's not the case.
	return checkOK, errwrap.Wrapf(err, "timeout after %.3f seconds", duration.Seconds())
}

// safeWatch wraps a call to res.Watch with a panic recovery so that a buggy
// resource or upstream library doesn't bring down the entire engine.
func safeWatch(ctx context.Context, res engine.Res) (err error) {
	defer func() {
		if r := recover(); r != nil {
			err = fmt.Errorf("panic in Watch: %+v", r)
		}
	}()
	return res.Watch(ctx)
}
