mesh-catalog/modules/messenger/cmd/messenger/takeover.go

package main

import (
	"sync"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/health"
)

// takeover is which seat's condition events the messenger hears: the controller's until the keeper's take-over key
// says otherwise (novox/hq ADR 0564 clause 9, issue 1952 R1). An absent key follows the controller.
type takeover struct {
	mu sync.Mutex
	t  health.Takeover
}

func newTakeover() *takeover {
	return &takeover{t: health.Takeover{V: health.TakeoverVersion, Follow: health.FollowController}}
}

func (f *takeover) seat() string { f.mu.Lock(); defer f.mu.Unlock(); return f.t.Seat() }

// set takes the key as a watch handed it and says whether the followed seat changed.
func (f *takeover) set(value []byte, found bool) (changed bool, err error) {
	t, err := health.DecodeTakeover(value, found)
	if err != nil {
		return false, err
	}
	f.mu.Lock()
	defer f.mu.Unlock()
	changed = f.t.Seat() != t.Seat()
	f.t = t
	return changed, nil
}

// followTakeover watches the take-over key for the life of the process, and asks the holder to read the followed
// seat's open conditions again when it changes. A key it cannot read leaves the seat it followed.
func followTakeover(f *takeover, h *Holder) {
	for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
		err := stdio.State(health.KeeperModule+"."+health.TakeoverState).Watch(health.TakeoverKey, func(c stdio.StateChange) error {
			changed, err := f.set(c.Value, c.Op != "delete" && len(c.Value) > 0)
			if err != nil {
				logf("[messenger] the take-over key is unreadable (%v); still hearing %s", err, f.seat())
				return nil
			}
			if changed {
				logf("[messenger] the take-over key now names %s", f.seat())
				h.AskSync()
			}
			return nil
		})
		if err == nil {
			return
		}
		logf("[messenger] not watching the take-over key yet (%v); hearing %s", err, f.seat())
		time.Sleep(wait)
	}
}