package main
import (
"log"
"sync"
"craftdesign.group/quarry/internal/api"
"craftdesign.group/quarry/internal/lease"
"craftdesign.group/quarry/internal/oltp"
)
// fleetLease is the lease code fleet-wide background jobs key on: the
// ones tied to no single store (CI status polling, the chat window
// sweep). A project code is hex, so it can never collide with this.
const fleetLease = "fleet"
// jobGate decides which node runs a background job, by reusing the
// store leases: a store's jobs run where its lease is held, fleet jobs
// where fleetLease is held. With one node that is every job, here.
// A nil gate, or one without a lease manager, runs everything.
type jobGate struct {
leases *lease.Manager
code func(repo string) (string, bool)
said sync.Map // lease code -> bool: the placement last logged
}
// store reports whether this node runs repo's jobs. A repository with
// no known code runs here, so its job reports what it finds exactly as
// it did before placement existed.
func (g *jobGate) store(repo string) bool {
if g == nil || g.leases == nil || g.code == nil {
return true
}
code, ok := g.code(repo)
if !ok {
return true
}
return g.held(code, repo)
}
// fleet reports whether this node runs the fleet-wide jobs.
func (g *jobGate) fleet() bool {
if g == nil || g.leases == nil {
return true
}
return g.held(fleetLease, "fleet jobs")
}
// held asks the lease and says so once each time the answer changes,
// never per tick: a skipped job on a non-holder is the design working,
// not a fault.
func (g *jobGate) held(code, what string) bool {
here := g.leases.Held(code)
prev, seen := g.said.Swap(code, here)
switch {
case !seen && here:
// The ordinary start: nothing moved.
case !seen, prev.(bool) != here:
if here {
log.Printf("jobs: %s run on this node now", what)
} else {
log.Printf("jobs: %s run on the node holding the lease, not this one", what)
}
}
return here
}
// heldSweeper narrows the auto-accept sweep to the stores this node
// runs jobs for; an accept anywhere else would only be refused by the
// lease and logged as a wait every sweep.
type heldSweeper struct {
api.AutoSweeper
jobs *jobGate
}
func (s heldSweeper) AutoAccepts() ([]oltp.AutoAccept, error) {
rows, err := s.AutoSweeper.AutoAccepts()
if err != nil {
return nil, err
}
mine := rows[:0:0]
for _, r := range rows {
if s.jobs.store(r.Repo) {
mine = append(mine, r)
}
}
return mine, nil
}