Files
mesh-delivery/modules/mesh-delivery-watchman/internal/watch/mirror.go
T

91 lines
2.7 KiB
Go

package watch
import (
"encoding/json"
"slices"
"strings"
"time"
)
// reservedBuildModules are the first parts of the builds bucket's keys that are not a module's.
var reservedBuildModules = []string{"manifest", "req", "id", "layout"}
// buildRecord is the part of the controller's build record (mesh-controller internal/chain Build) the watchman reads:
// a record in state asked is a request that no outcome has answered.
type buildRecord struct {
ID string `json:"id"`
State string `json:"state"`
Module string `json:"module"`
Asked time.Time `json:"asked"`
At time.Time `json:"at"`
}
// Build is a change in the builds bucket. A record `<module>.build.<id>` in state asked, with no outcome, is mirrored;
// when its outcome is written it has moved. No other key of the bucket is mirrored.
func (w *Watchman) Build(key string, value []byte, at time.Time, op Op) {
w.mu.Lock()
defer w.mu.Unlock()
parts := strings.Split(key, ".")
if len(parts) != 3 || parts[1] != "build" || slices.Contains(reservedBuildModules, parts[0]) || op == Expired {
return
}
var rec buildRecord
if op == Delete || json.Unmarshal(value, &rec) != nil || rec.State != "asked" {
w.moved(BuildsBucket, key)
return
}
module := rec.Module
if module == "" {
module = parts[0]
}
since := rec.Asked
if since.IsZero() {
since = rec.At
}
if since.IsZero() {
since = at
}
w.mirror(BuildsBucket, key, describe("the build of %s to be answered", module), rec.ID, since, 0)
}
// checkRecord is the part of a running check's key the watchman reads: the head being checked and when the check
// started (mesh-controller issue 1155).
type checkRecord struct {
Head string `json:"head"`
Started time.Time `json:"started"`
}
// Check is a change in the checks-running bucket. A key is a check running, mirrored from its start; a builder's
// renewal of the key changes nothing, a new head or start is a new check, and a deleted key is the check ended. An
// aged-out key is the bucket's own expiry of a builder that died: the pulse is left to run out.
func (w *Watchman) Check(key string, value []byte, at time.Time, op Op) {
w.mu.Lock()
defer w.mu.Unlock()
switch op {
case Expired:
return
case Delete:
w.moved(ChecksBucket, key)
return
}
var rec checkRecord
_ = json.Unmarshal(value, &rec)
since := rec.Started
if since.IsZero() {
since = at
if cur := w.pulseOf(ChecksBucket, key); cur != nil && cur.head == rec.Head && !cur.seeded {
since = cur.since
}
}
w.mirror(ChecksBucket, key, describe("the check of %s to end", key), rec.Head, since, 0)
}
func (w *Watchman) pulseOf(bucket, key string) *pulse {
for _, p := range w.pulses {
if p.bucket == bucket && p.key == key {
return p
}
}
return nil
}