feat: 实现 Sense Brain 控制连接器 (#152)
This commit is contained in:
QiuSW
@@ -0,0 +1,118 @@
|
||||
package brain_control
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"time"
|
||||
|
||||
mi "git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/machine_identity"
|
||||
)
|
||||
|
||||
const (
|
||||
sourceConfigPath = "/machine/v1/source-config"
|
||||
runtimeStatusPath = "/machine/v1/runtime-status"
|
||||
)
|
||||
|
||||
var requestIDPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:-]{15,127}$`)
|
||||
|
||||
type SendResponse struct {
|
||||
StatusCode int
|
||||
Body []byte
|
||||
CorrelationID string
|
||||
}
|
||||
type SendFunc func(authorization string, body []byte, correlationID string, timeout time.Duration) (SendResponse, error)
|
||||
|
||||
type ConfigSender struct {
|
||||
Signer mi.Signer
|
||||
Send SendFunc
|
||||
Enabled bool
|
||||
Timeout time.Duration
|
||||
MaxAttempts int
|
||||
Sleep func(time.Duration)
|
||||
}
|
||||
|
||||
func (s ConfigSender) Publish(config SourceConfig, correlationID string) (SendResponse, error) {
|
||||
if !s.Enabled {
|
||||
return SendResponse{}, errors.New("CONNECTOR_DISABLED")
|
||||
}
|
||||
if s.Send == nil || !requestIDPattern.MatchString(correlationID) {
|
||||
return SendResponse{}, errors.New("invalid connector configuration")
|
||||
}
|
||||
if s.Timeout <= 0 {
|
||||
s.Timeout = 5 * time.Second
|
||||
}
|
||||
if s.MaxAttempts == 0 {
|
||||
s.MaxAttempts = 4
|
||||
}
|
||||
if s.MaxAttempts < 1 {
|
||||
return SendResponse{}, errors.New("invalid connector retry policy")
|
||||
}
|
||||
if s.Sleep == nil {
|
||||
s.Sleep = time.Sleep
|
||||
}
|
||||
if err := ValidateSourceConfig(config); err != nil {
|
||||
return SendResponse{}, err
|
||||
}
|
||||
body, err := MarshalSourceConfig(config)
|
||||
if err != nil {
|
||||
return SendResponse{}, err
|
||||
}
|
||||
var last error
|
||||
for attempt := 0; attempt < s.MaxAttempts; attempt++ {
|
||||
token, mintErr := s.Signer.Mint("yovision-brain", []string{"source-config:write"}, "POST", sourceConfigPath, body)
|
||||
if mintErr != nil {
|
||||
return SendResponse{}, mintErr
|
||||
}
|
||||
response, sendErr := s.Send("Bearer "+token, body, correlationID, s.Timeout)
|
||||
if sendErr == nil && response.StatusCode >= 200 && response.StatusCode < 300 {
|
||||
return response, nil
|
||||
}
|
||||
if sendErr == nil && response.StatusCode < 500 {
|
||||
return SendResponse{}, fmt.Errorf("source config rejected: %d", response.StatusCode)
|
||||
}
|
||||
if sendErr != nil {
|
||||
last = sendErr
|
||||
} else {
|
||||
last = fmt.Errorf("source config remote status: %d", response.StatusCode)
|
||||
}
|
||||
if attempt+1 < s.MaxAttempts {
|
||||
delay := time.Second << attempt
|
||||
if delay > 30*time.Second {
|
||||
delay = 30 * time.Second
|
||||
}
|
||||
s.Sleep(delay)
|
||||
}
|
||||
}
|
||||
return SendResponse{}, fmt.Errorf("source config delivery exhausted: %w", last)
|
||||
}
|
||||
|
||||
type RuntimeStatusEndpoint struct {
|
||||
Verifier mi.Verifier
|
||||
Store ProjectionStore
|
||||
Enabled bool
|
||||
MaxBodyBytes int
|
||||
}
|
||||
|
||||
func (e RuntimeStatusEndpoint) Receive(authorization string, body []byte, correlationID string, expected map[string]int64) (ProjectionView, error) {
|
||||
if !e.Enabled {
|
||||
return ProjectionView{}, errors.New("CONNECTOR_DISABLED")
|
||||
}
|
||||
if !requestIDPattern.MatchString(correlationID) {
|
||||
return ProjectionView{}, errors.New("INVALID_CORRELATION_ID")
|
||||
}
|
||||
if e.MaxBodyBytes == 0 {
|
||||
e.MaxBodyBytes = 10 * 1024 * 1024
|
||||
}
|
||||
if len(body) > e.MaxBodyBytes {
|
||||
return ProjectionView{}, errors.New("REQUEST_TOO_LARGE")
|
||||
}
|
||||
token, err := mi.BearerToken(authorization)
|
||||
if err != nil {
|
||||
return ProjectionView{}, err
|
||||
}
|
||||
if _, err = e.Verifier.Verify(token, "yovision-sense", "runtime-status:write", "POST", runtimeStatusPath, body); err != nil {
|
||||
return ProjectionView{}, err
|
||||
}
|
||||
return e.Store.Ingest(body, expected)
|
||||
}
|
||||
@@ -0,0 +1,124 @@
|
||||
package brain_control
|
||||
|
||||
import (
|
||||
"crypto/ed25519"
|
||||
"crypto/rand"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
mi "git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/machine_identity"
|
||||
"gorm.io/driver/sqlite"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func memoryDB(t *testing.T, name string) *gorm.DB {
|
||||
t.Helper()
|
||||
db, err := gorm.Open(sqlite.Open("file:"+name+"?mode=memory&cache=shared"), &gorm.Config{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err = db.AutoMigrate(&ReplayToken{}, &RuntimeProjection{}, &SourceRevision{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return db
|
||||
}
|
||||
|
||||
func TestSourceMapperAndRecalibration(t *testing.T) {
|
||||
now := time.Date(2026, 8, 31, 0, 0, 0, 0, time.UTC)
|
||||
facts := SourceFacts{ConfigID: "gate-primary", SiteID: "site-east", LogicalDeviceID: "camera-1", MediaPath: "site/camera/main", Revision: 1, PublishedAt: now, EffectiveAt: now, Profile: Profile{ID: "main", Width: 1920, Height: 1080, Encoding: "h264", FrameRate: 25}, RuleSetVersion: "rules-1", Areas: []AreaRule{{ID: "danger", Version: 1, Kind: "danger_area", Enabled: true, Points: []Point{{.1, .1}, {.8, .1}, {.5, .8}}}}}
|
||||
config, err := MapSourceConfig(facts)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err = ValidateSourceConfig(config); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
facts.Revision = 2
|
||||
facts.NeedsRecalibration = true
|
||||
config, err = MapSourceConfig(facts)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if config.RuleSet.State != "recalibration_required" || config.RuleSet.Areas[0].Enabled {
|
||||
t.Fatal("recalibration did not disable rules")
|
||||
}
|
||||
}
|
||||
|
||||
func TestReplayAtomicAndRestart(t *testing.T) {
|
||||
db := memoryDB(t, "replay-package")
|
||||
pub, priv, _ := ed25519.GenerateKey(rand.Reader)
|
||||
now := time.Date(2026, 8, 31, 0, 0, 0, 0, time.UTC)
|
||||
registry, _ := mi.NewRegistry(mi.KeyRecord{Principal: "yv:brain:east", KeyID: "brain-key-01", PublicKey: pub, Audience: "yovision-sense", Scopes: []string{"runtime-status:write"}, Enabled: true})
|
||||
signer := mi.Signer{Principal: "yv:brain:east", KeyID: "brain-key-01", PrivateKey: priv, Now: func() time.Time { return now }}
|
||||
body := []byte("{}")
|
||||
token, _ := signer.Mint("yovision-sense", []string{"runtime-status:write"}, "POST", "/machine/v1/runtime-status", body)
|
||||
accepted := 0
|
||||
var mu sync.Mutex
|
||||
var wg sync.WaitGroup
|
||||
for range 8 {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
v := mi.Verifier{Registry: registry, Replay: GORMReplayStore{DB: db}, Now: func() time.Time { return now }}
|
||||
if _, err := v.Verify(token, "yovision-sense", "runtime-status:write", "POST", "/machine/v1/runtime-status", body); err == nil {
|
||||
mu.Lock()
|
||||
accepted++
|
||||
mu.Unlock()
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
if accepted != 1 {
|
||||
t.Fatalf("accepted %d", accepted)
|
||||
}
|
||||
v := mi.Verifier{Registry: registry, Replay: GORMReplayStore{DB: db}, Now: func() time.Time { return now }}
|
||||
if _, err := v.Verify(token, "yovision-sense", "runtime-status:write", "POST", "/machine/v1/runtime-status", body); err == nil {
|
||||
t.Fatal("restart replay accepted")
|
||||
}
|
||||
}
|
||||
|
||||
func TestProjectionStaleRecoveryAndMismatch(t *testing.T) {
|
||||
db := memoryDB(t, "projection-package")
|
||||
now := time.Date(2026, 8, 31, 0, 0, 0, 0, time.UTC)
|
||||
store := ProjectionStore{DB: db, Clock: func() time.Time { return now }, StaleAfter: 90 * time.Second}
|
||||
raw := runtimeFixture("018f4d6a-8d1b-4a25-8b37-9085f9c0d101", 1, now, 2)
|
||||
view, err := store.Ingest(raw, map[string]int64{"gate": 3})
|
||||
if err != nil || !view.RevisionMismatch {
|
||||
t.Fatalf("view %+v err %v", view, err)
|
||||
}
|
||||
now = now.Add(91 * time.Second)
|
||||
view, _ = store.View("brain-east-01")
|
||||
if !view.Offline {
|
||||
t.Fatal("not offline")
|
||||
}
|
||||
raw = runtimeFixture("018f4d6a-8d1b-4a25-8b37-9085f9c0d102", 2, now, 3)
|
||||
view, err = store.Ingest(raw, map[string]int64{"gate": 3})
|
||||
if err != nil || !view.Recovered {
|
||||
t.Fatalf("recovery %+v err %v", view, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConnectorRejectsUnsafeRequestIDs(t *testing.T) {
|
||||
for _, value := range []string{"short", "0123456789abcde\n", "0123456789abcde!", strings.Repeat("a", 129)} {
|
||||
sender := ConfigSender{Enabled: true, Send: func(string, []byte, string, time.Duration) (SendResponse, error) {
|
||||
t.Fatal("unsafe request id reached transport")
|
||||
return SendResponse{}, nil
|
||||
}}
|
||||
if _, err := sender.Publish(SourceConfig{}, value); err == nil {
|
||||
t.Fatalf("sender accepted request id %q", value)
|
||||
}
|
||||
endpoint := RuntimeStatusEndpoint{Enabled: true}
|
||||
if _, err := endpoint.Receive("Bearer ignored", nil, value, nil); err == nil || err.Error() != "INVALID_CORRELATION_ID" {
|
||||
t.Fatalf("endpoint accepted request id %q: %v", value, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func runtimeFixture(id string, seq int64, observed time.Time, revision int64) []byte {
|
||||
value := map[string]any{"schema_version": RuntimeStatusVersion, "status_id": id, "brain_instance_ref": "brain-east-01", "sequence": seq, "observed_at": observed.Format(time.RFC3339), "runtime": map[string]any{"state": "running", "version": "1.0.0", "started_at": observed.Format(time.RFC3339)}, "model": map[string]any{"model_ref": "people", "version": "1"}, "configurations": []any{map[string]any{"config_id": "gate", "apply_state": "applied", "applied_revision": revision, "error_code": nil}}, "health": map[string]any{"overall": "healthy", "error_codes": []any{}, "metrics": map[string]any{"load_percent": 1.0, "queue_depth": 0, "latency_ms": 1.0}}, "inputs": []any{}}
|
||||
raw, _ := json.Marshal(value)
|
||||
return raw
|
||||
}
|
||||
@@ -0,0 +1,135 @@
|
||||
package brain_control
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
var stableID = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._~-]{0,127}$`)
|
||||
|
||||
func MapSourceConfig(f SourceFacts) (SourceConfig, error) {
|
||||
if !stableID.MatchString(f.ConfigID) || !stableID.MatchString(f.SiteID) || !stableID.MatchString(f.LogicalDeviceID) || !stableID.MatchString(f.Profile.ID) || f.Revision < 1 || f.Profile.Width < 1 || f.Profile.Height < 1 || f.Profile.FrameRate <= 0 {
|
||||
return SourceConfig{}, errors.New("invalid source configuration facts")
|
||||
}
|
||||
encoding := strings.ToUpper(f.Profile.Encoding)
|
||||
if encoding != "H264" && encoding != "H265" && encoding != "MJPEG" {
|
||||
return SourceConfig{}, errors.New("unsupported profile encoding")
|
||||
}
|
||||
if f.PublishedAt.IsZero() || f.EffectiveAt.Before(f.PublishedAt) {
|
||||
return SourceConfig{}, errors.New("invalid source configuration time")
|
||||
}
|
||||
if !stableID.MatchString(f.RuleSetVersion) {
|
||||
return SourceConfig{}, errors.New("invalid rule set version")
|
||||
}
|
||||
if strings.ContainsAny(f.MediaPath, "?#@\\") || strings.Contains(f.MediaPath, "://") || f.MediaPath == "" {
|
||||
return SourceConfig{}, errors.New("media path must be opaque and credential-free")
|
||||
}
|
||||
if len(f.Areas) > 1024 || len(f.DirectionalLines) > 1024 {
|
||||
return SourceConfig{}, errors.New("too many rules")
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
for _, a := range f.Areas {
|
||||
if !stableID.MatchString(a.ID) || seen[a.ID] || a.Version < 1 || a.Kind != "danger_area" || len(a.Points) < 3 || len(a.Points) > 256 || !validPoints(a.Points) || polygonArea(a.Points) == 0 {
|
||||
return SourceConfig{}, errors.New("invalid area rule")
|
||||
}
|
||||
seen[a.ID] = true
|
||||
}
|
||||
for _, l := range f.DirectionalLines {
|
||||
if !stableID.MatchString(l.ID) || seen[l.ID] || l.Version < 1 || l.Kind != "directional_line" || (l.TriggerDirection != "left_to_right" && l.TriggerDirection != "right_to_left") || !validPoints([]Point{l.Start, l.End}) || l.Start == l.End {
|
||||
return SourceConfig{}, errors.New("invalid directional line rule")
|
||||
}
|
||||
seen[l.ID] = true
|
||||
}
|
||||
var out SourceConfig
|
||||
out.SchemaVersion, out.ConfigID, out.Revision = SourceConfigVersion, f.ConfigID, f.Revision
|
||||
out.PublishedAt, out.EffectiveAt = f.PublishedAt.UTC(), f.EffectiveAt.UTC()
|
||||
out.Site.ID, out.LogicalDevice.ID = f.SiteID, f.LogicalDeviceID
|
||||
out.Profile = f.Profile
|
||||
out.Profile.Encoding = encoding
|
||||
out.Media.Ref, out.Media.Transport = "media:"+strings.TrimPrefix(f.MediaPath, "/"), "rtsp"
|
||||
out.RuleSet.Version = f.RuleSetVersion
|
||||
out.RuleSet.State = "active"
|
||||
if f.Disabled {
|
||||
out.RuleSet.State = "disabled"
|
||||
}
|
||||
if f.NeedsRecalibration {
|
||||
out.RuleSet.State = "recalibration_required"
|
||||
}
|
||||
out.RuleSet.ProfileBinding.ProfileID, out.RuleSet.ProfileBinding.Width, out.RuleSet.ProfileBinding.Height = f.Profile.ID, f.Profile.Width, f.Profile.Height
|
||||
out.RuleSet.Areas = append([]AreaRule(nil), f.Areas...)
|
||||
out.RuleSet.DirectionalLines = append([]DirectionalLineRule(nil), f.DirectionalLines...)
|
||||
if out.RuleSet.State != "active" {
|
||||
for i := range out.RuleSet.Areas {
|
||||
out.RuleSet.Areas[i].Enabled = false
|
||||
}
|
||||
for i := range out.RuleSet.DirectionalLines {
|
||||
out.RuleSet.DirectionalLines[i].Enabled = false
|
||||
}
|
||||
}
|
||||
digest, err := sourceDigest(out)
|
||||
if err != nil {
|
||||
return SourceConfig{}, err
|
||||
}
|
||||
out.Integrity.Algorithm, out.Integrity.Value = "sha256", digest
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func MarshalSourceConfig(config SourceConfig) ([]byte, error) { return json.Marshal(config) }
|
||||
|
||||
func sourceDigest(config SourceConfig) (string, error) {
|
||||
raw, err := json.Marshal(config)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
var value map[string]any
|
||||
if err = json.Unmarshal(raw, &value); err != nil {
|
||||
return "", err
|
||||
}
|
||||
delete(value, "integrity")
|
||||
canonical, err := json.Marshal(value)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
sum := sha256.Sum256(canonical)
|
||||
return hex.EncodeToString(sum[:]), nil
|
||||
}
|
||||
|
||||
func validPoints(points []Point) bool {
|
||||
for _, p := range points {
|
||||
if p.X < 0 || p.X > 1 || p.Y < 0 || p.Y > 1 {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
func polygonArea(p []Point) float64 {
|
||||
var a float64
|
||||
for i := range p {
|
||||
n := p[(i+1)%len(p)]
|
||||
a += p[i].X*n.Y - n.X*p[i].Y
|
||||
}
|
||||
if a < 0 {
|
||||
a = -a
|
||||
}
|
||||
return a / 2
|
||||
}
|
||||
|
||||
func ValidateSourceConfig(config SourceConfig) error {
|
||||
if config.SchemaVersion != SourceConfigVersion {
|
||||
return fmt.Errorf("unsupported source config version")
|
||||
}
|
||||
digest, err := sourceDigest(config)
|
||||
if err != nil || config.Integrity.Algorithm != "sha256" || digest != config.Integrity.Value {
|
||||
return errors.New("source config integrity mismatch")
|
||||
}
|
||||
_, err = MapSourceConfig(SourceFacts{ConfigID: config.ConfigID, SiteID: config.Site.ID, LogicalDeviceID: config.LogicalDevice.ID, MediaPath: strings.TrimPrefix(config.Media.Ref, "media:"), Revision: config.Revision, PublishedAt: config.PublishedAt, EffectiveAt: config.EffectiveAt, Profile: config.Profile, RuleSetVersion: config.RuleSet.Version, Disabled: config.RuleSet.State == "disabled", NeedsRecalibration: config.RuleSet.State == "recalibration_required", Areas: config.RuleSet.Areas, DirectionalLines: config.RuleSet.DirectionalLines})
|
||||
return err
|
||||
}
|
||||
|
||||
func UTCNow() time.Time { return time.Now().UTC() }
|
||||
@@ -0,0 +1,122 @@
|
||||
package brain_control
|
||||
|
||||
import "time"
|
||||
|
||||
const (
|
||||
SourceConfigVersion = "yovision.source-config/v1"
|
||||
RuntimeStatusVersion = "yovision.runtime-status/v1"
|
||||
)
|
||||
|
||||
type Point struct {
|
||||
X float64 `json:"x"`
|
||||
Y float64 `json:"y"`
|
||||
}
|
||||
type Profile struct {
|
||||
ID string `json:"id"`
|
||||
Width int `json:"width"`
|
||||
Height int `json:"height"`
|
||||
Encoding string `json:"encoding"`
|
||||
FrameRate float64 `json:"frame_rate"`
|
||||
}
|
||||
type AreaRule struct {
|
||||
ID string `json:"id"`
|
||||
Version int64 `json:"version"`
|
||||
Kind string `json:"kind"`
|
||||
Enabled bool `json:"enabled"`
|
||||
Points []Point `json:"points"`
|
||||
}
|
||||
type DirectionalLineRule struct {
|
||||
ID string `json:"id"`
|
||||
Version int64 `json:"version"`
|
||||
Kind string `json:"kind"`
|
||||
Enabled bool `json:"enabled"`
|
||||
Start Point `json:"start"`
|
||||
End Point `json:"end"`
|
||||
TriggerDirection string `json:"trigger_direction"`
|
||||
}
|
||||
|
||||
type SourceConfig struct {
|
||||
SchemaVersion string `json:"schema_version"`
|
||||
ConfigID string `json:"config_id"`
|
||||
Revision int64 `json:"revision"`
|
||||
PublishedAt time.Time `json:"published_at"`
|
||||
EffectiveAt time.Time `json:"effective_at"`
|
||||
Site struct {
|
||||
ID string `json:"id"`
|
||||
} `json:"site"`
|
||||
LogicalDevice struct {
|
||||
ID string `json:"id"`
|
||||
} `json:"logical_device"`
|
||||
Profile Profile `json:"profile"`
|
||||
Media struct {
|
||||
Ref string `json:"ref"`
|
||||
Transport string `json:"transport"`
|
||||
} `json:"media"`
|
||||
RuleSet struct {
|
||||
Version string `json:"version"`
|
||||
State string `json:"state"`
|
||||
ProfileBinding struct {
|
||||
ProfileID string `json:"profile_id"`
|
||||
Width int `json:"width"`
|
||||
Height int `json:"height"`
|
||||
} `json:"profile_binding"`
|
||||
Areas []AreaRule `json:"areas"`
|
||||
DirectionalLines []DirectionalLineRule `json:"directional_lines"`
|
||||
} `json:"rule_set"`
|
||||
Integrity struct {
|
||||
Algorithm string `json:"algorithm"`
|
||||
Value string `json:"value"`
|
||||
} `json:"integrity"`
|
||||
}
|
||||
|
||||
// SourceFacts is an explicit, credential-free boundary DTO. Callers map their
|
||||
// GORM entities into it; database models are never serialized as a contract.
|
||||
type SourceFacts struct {
|
||||
ConfigID, SiteID, LogicalDeviceID, MediaRouteID, MediaPath string
|
||||
Revision int64
|
||||
PublishedAt, EffectiveAt time.Time
|
||||
Profile Profile
|
||||
RuleSetVersion string
|
||||
Disabled, NeedsRecalibration bool
|
||||
Areas []AreaRule
|
||||
DirectionalLines []DirectionalLineRule
|
||||
}
|
||||
|
||||
type ReplayToken struct {
|
||||
Principal string `gorm:"size:128;primaryKey"`
|
||||
TokenID string `gorm:"size:96;primaryKey"`
|
||||
ExpiresAt time.Time `gorm:"not null;index"`
|
||||
CreatedAt time.Time `gorm:"not null"`
|
||||
}
|
||||
|
||||
func (ReplayToken) TableName() string { return "sense_brain_runtime_replay_tokens" }
|
||||
|
||||
type RuntimeProjection struct {
|
||||
BrainInstanceRef string `gorm:"size:128;primaryKey"`
|
||||
StatusID string `gorm:"size:36;not null;uniqueIndex"`
|
||||
Sequence int64 `gorm:"not null"`
|
||||
ObservedAt time.Time `gorm:"not null;index"`
|
||||
ReceivedAt time.Time `gorm:"not null"`
|
||||
RuntimeState string `gorm:"size:32;not null"`
|
||||
RuntimeVersion string `gorm:"size:64;not null"`
|
||||
ModelRef string `gorm:"size:128;not null"`
|
||||
ModelVersion string `gorm:"size:64;not null"`
|
||||
HealthOverall string `gorm:"size:32;not null"`
|
||||
ExpectedRevisionsJSON string `gorm:"type:jsonb;not null"`
|
||||
ConfigurationsJSON string `gorm:"type:jsonb;not null"`
|
||||
HealthJSON string `gorm:"type:jsonb;not null"`
|
||||
InputsJSON string `gorm:"type:jsonb;not null"`
|
||||
WasOffline bool `gorm:"not null;default:false"`
|
||||
CreatedAt time.Time
|
||||
UpdatedAt time.Time
|
||||
}
|
||||
|
||||
func (RuntimeProjection) TableName() string { return "sense_brain_runtime_projections" }
|
||||
|
||||
type SourceRevision struct {
|
||||
ConfigID string `gorm:"size:128;primaryKey"`
|
||||
Revision int64 `gorm:"not null"`
|
||||
UpdatedAt time.Time
|
||||
}
|
||||
|
||||
func (SourceRevision) TableName() string { return "sense_brain_source_revisions" }
|
||||
@@ -0,0 +1,145 @@
|
||||
package brain_control
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"regexp"
|
||||
"time"
|
||||
)
|
||||
|
||||
type runtimeStatus struct {
|
||||
SchemaVersion string `json:"schema_version"`
|
||||
StatusID string `json:"status_id"`
|
||||
BrainInstanceRef string `json:"brain_instance_ref"`
|
||||
Sequence int64 `json:"sequence"`
|
||||
ObservedAt time.Time `json:"observed_at"`
|
||||
Runtime struct {
|
||||
State string `json:"state"`
|
||||
Version string `json:"version"`
|
||||
StartedAt *time.Time `json:"started_at"`
|
||||
} `json:"runtime"`
|
||||
Model struct {
|
||||
ModelRef string `json:"model_ref"`
|
||||
Version string `json:"version"`
|
||||
} `json:"model"`
|
||||
Configurations []configurationStatus `json:"configurations"`
|
||||
Health healthStatus `json:"health"`
|
||||
Inputs []inputStatus `json:"inputs"`
|
||||
}
|
||||
type configurationStatus struct {
|
||||
ConfigID string `json:"config_id"`
|
||||
ApplyState string `json:"apply_state"`
|
||||
AppliedRevision *int64 `json:"applied_revision"`
|
||||
ErrorCode *string `json:"error_code"`
|
||||
}
|
||||
type metrics struct {
|
||||
LoadPercent float64 `json:"load_percent"`
|
||||
QueueDepth int64 `json:"queue_depth"`
|
||||
LatencyMS float64 `json:"latency_ms"`
|
||||
}
|
||||
type healthStatus struct {
|
||||
Overall string `json:"overall"`
|
||||
ErrorCodes []string `json:"error_codes"`
|
||||
Metrics metrics `json:"metrics"`
|
||||
}
|
||||
type inputStatus struct {
|
||||
InputRef string `json:"input_ref"`
|
||||
State string `json:"state"`
|
||||
ErrorCodes []string `json:"error_codes"`
|
||||
Metrics metrics `json:"metrics"`
|
||||
}
|
||||
|
||||
var uuid4 = regexp.MustCompile(`^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$`)
|
||||
var errCode = regexp.MustCompile(`^[A-Z][A-Z0-9_]{2,63}$`)
|
||||
|
||||
func parseRuntimeStatus(raw []byte) (runtimeStatus, error) {
|
||||
var s runtimeStatus
|
||||
d := json.NewDecoder(bytes.NewReader(raw))
|
||||
d.DisallowUnknownFields()
|
||||
if err := d.Decode(&s); err != nil {
|
||||
return s, errors.New("CONFIG_INVALID")
|
||||
}
|
||||
if err := d.Decode(&struct{}{}); !errors.Is(err, io.EOF) {
|
||||
return s, errors.New("CONFIG_INVALID")
|
||||
}
|
||||
if s.SchemaVersion != RuntimeStatusVersion {
|
||||
return s, errors.New("UNSUPPORTED_SCHEMA_VERSION")
|
||||
}
|
||||
if !uuid4.MatchString(s.StatusID) || !stableID.MatchString(s.BrainInstanceRef) || s.Sequence < 0 || s.ObservedAt.IsZero() || !validState(s.Runtime.State) || s.Runtime.Version == "" || !stableID.MatchString(s.Model.ModelRef) || s.Model.Version == "" || len(s.Configurations) > 4096 || len(s.Inputs) > 4096 || !validHealth(s.Health) {
|
||||
return s, errors.New("CONFIG_INVALID")
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
for _, c := range s.Configurations {
|
||||
if !stableID.MatchString(c.ConfigID) || seen[c.ConfigID] || !validApply(c) {
|
||||
return s, errors.New("CONFIG_INVALID")
|
||||
}
|
||||
seen[c.ConfigID] = true
|
||||
}
|
||||
for _, i := range s.Inputs {
|
||||
if !stableID.MatchString(i.InputRef) || !validState(i.State) || !validMetrics(i.Metrics) || !validCodes(i.ErrorCodes, 16) {
|
||||
return s, errors.New("CONFIG_INVALID")
|
||||
}
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
func validState(v string) bool {
|
||||
switch v {
|
||||
case "unconfigured", "starting", "running", "degraded", "failed", "stopped":
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
func validApply(c configurationStatus) bool {
|
||||
switch c.ApplyState {
|
||||
case "not_configured":
|
||||
return c.AppliedRevision == nil
|
||||
case "applying":
|
||||
return true
|
||||
case "applied":
|
||||
return c.AppliedRevision != nil && *c.AppliedRevision >= 1
|
||||
case "rejected":
|
||||
return c.ErrorCode != nil && errCode.MatchString(*c.ErrorCode)
|
||||
}
|
||||
return false
|
||||
}
|
||||
func validMetrics(m metrics) bool {
|
||||
return m.LoadPercent >= 0 && m.LoadPercent <= 100 && m.QueueDepth >= 0 && m.LatencyMS >= 0
|
||||
}
|
||||
func validCodes(v []string, n int) bool {
|
||||
if len(v) > n {
|
||||
return false
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
for _, x := range v {
|
||||
if seen[x] || !errCode.MatchString(x) {
|
||||
return false
|
||||
}
|
||||
seen[x] = true
|
||||
}
|
||||
return true
|
||||
}
|
||||
func validHealth(h healthStatus) bool {
|
||||
return (h.Overall == "healthy" || h.Overall == "degraded" || h.Overall == "unhealthy") && validCodes(h.ErrorCodes, 32) && validMetrics(h.Metrics)
|
||||
}
|
||||
func validRuntimeTransition(from, to string) bool {
|
||||
if from == to {
|
||||
return true
|
||||
}
|
||||
switch from {
|
||||
case "unconfigured":
|
||||
return to == "starting" || to == "stopped"
|
||||
case "starting":
|
||||
return to == "running" || to == "degraded" || to == "failed" || to == "stopped"
|
||||
case "running":
|
||||
return to == "degraded" || to == "failed" || to == "stopped"
|
||||
case "degraded":
|
||||
return to == "running" || to == "failed" || to == "stopped"
|
||||
case "failed":
|
||||
return to == "starting" || to == "stopped"
|
||||
case "stopped":
|
||||
return to == "starting"
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -0,0 +1,159 @@
|
||||
package brain_control
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
)
|
||||
|
||||
type GORMReplayStore struct{ DB *gorm.DB }
|
||||
|
||||
func (s GORMReplayStore) Consume(principal, tokenID string, expiresAt, now time.Time) bool {
|
||||
if s.DB == nil || principal == "" || tokenID == "" || !expiresAt.After(now) {
|
||||
return false
|
||||
}
|
||||
return s.DB.Transaction(func(tx *gorm.DB) error {
|
||||
if err := tx.Where("expires_at <= ?", now).Delete(&ReplayToken{}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
result := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&ReplayToken{Principal: principal, TokenID: tokenID, ExpiresAt: expiresAt, CreatedAt: now})
|
||||
if result.Error != nil {
|
||||
return result.Error
|
||||
}
|
||||
if result.RowsAffected != 1 {
|
||||
return errors.New("replayed")
|
||||
}
|
||||
return nil
|
||||
}) == nil
|
||||
}
|
||||
|
||||
type RevisionStore struct{ DB *gorm.DB }
|
||||
|
||||
func (s RevisionStore) Next(configID string) (int64, error) {
|
||||
if s.DB == nil || !stableID.MatchString(configID) {
|
||||
return 0, errors.New("invalid revision store")
|
||||
}
|
||||
var next int64
|
||||
err := s.DB.Transaction(func(tx *gorm.DB) error {
|
||||
var row SourceRevision
|
||||
err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("config_id = ?", configID).First(&row).Error
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
row = SourceRevision{ConfigID: configID, Revision: 1}
|
||||
if err = tx.Create(&row).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
next = 1
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
row.Revision++
|
||||
next = row.Revision
|
||||
return tx.Save(&row).Error
|
||||
})
|
||||
return next, err
|
||||
}
|
||||
|
||||
type ProjectionStore struct {
|
||||
DB *gorm.DB
|
||||
Clock func() time.Time
|
||||
StaleAfter time.Duration
|
||||
FutureSkew time.Duration
|
||||
}
|
||||
type ProjectionView struct {
|
||||
Projection RuntimeProjection
|
||||
Offline, Stale, Recovered, RevisionMismatch bool
|
||||
}
|
||||
|
||||
func (s ProjectionStore) Ingest(raw []byte, expected map[string]int64) (ProjectionView, error) {
|
||||
if s.DB == nil {
|
||||
return ProjectionView{}, errors.New("projection database required")
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
if s.Clock != nil {
|
||||
now = s.Clock().UTC()
|
||||
}
|
||||
if s.StaleAfter == 0 {
|
||||
s.StaleAfter = 90 * time.Second
|
||||
}
|
||||
if s.FutureSkew == 0 {
|
||||
s.FutureSkew = 30 * time.Second
|
||||
}
|
||||
status, err := parseRuntimeStatus(raw)
|
||||
if err != nil {
|
||||
return ProjectionView{}, err
|
||||
}
|
||||
if status.ObservedAt.After(now.Add(s.FutureSkew)) {
|
||||
return ProjectionView{}, errors.New("FUTURE_OBSERVATION")
|
||||
}
|
||||
var view ProjectionView
|
||||
err = s.DB.Transaction(func(tx *gorm.DB) error {
|
||||
var old RuntimeProjection
|
||||
find := tx.Where("brain_instance_ref = ?", status.BrainInstanceRef).First(&old).Error
|
||||
if find == nil {
|
||||
if old.StatusID == status.StatusID {
|
||||
view.Projection = old
|
||||
return nil
|
||||
}
|
||||
if status.Sequence <= old.Sequence {
|
||||
return errors.New("OUT_OF_ORDER_STATUS")
|
||||
}
|
||||
if !validRuntimeTransition(old.RuntimeState, status.Runtime.State) {
|
||||
return errors.New("INVALID_STATUS_TRANSITION")
|
||||
}
|
||||
}
|
||||
if find != nil && !errors.Is(find, gorm.ErrRecordNotFound) {
|
||||
return find
|
||||
}
|
||||
expectedJSON, _ := json.Marshal(expected)
|
||||
configs, _ := json.Marshal(status.Configurations)
|
||||
health, _ := json.Marshal(status.Health)
|
||||
inputs, _ := json.Marshal(status.Inputs)
|
||||
p := RuntimeProjection{BrainInstanceRef: status.BrainInstanceRef, StatusID: status.StatusID, Sequence: status.Sequence, ObservedAt: status.ObservedAt, ReceivedAt: now, RuntimeState: status.Runtime.State, RuntimeVersion: status.Runtime.Version, ModelRef: status.Model.ModelRef, ModelVersion: status.Model.Version, HealthOverall: status.Health.Overall, ExpectedRevisionsJSON: string(expectedJSON), ConfigurationsJSON: string(configs), HealthJSON: string(health), InputsJSON: string(inputs), WasOffline: find == nil && now.Sub(old.ObservedAt) > s.StaleAfter}
|
||||
if find == nil {
|
||||
p.CreatedAt = old.CreatedAt
|
||||
view.Recovered = p.WasOffline
|
||||
}
|
||||
if err := tx.Save(&p).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
view.Projection = p
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return ProjectionView{}, err
|
||||
}
|
||||
view.Stale = now.Sub(view.Projection.ObservedAt) > s.StaleAfter
|
||||
view.Offline = view.Stale
|
||||
applied := make(map[string]*int64, len(status.Configurations))
|
||||
for _, c := range status.Configurations {
|
||||
applied[c.ConfigID] = c.AppliedRevision
|
||||
}
|
||||
for configID, want := range expected {
|
||||
got, ok := applied[configID]
|
||||
if !ok || got == nil || *got != want {
|
||||
view.RevisionMismatch = true
|
||||
}
|
||||
}
|
||||
return view, nil
|
||||
}
|
||||
|
||||
func (s ProjectionStore) View(instance string) (ProjectionView, error) {
|
||||
var p RuntimeProjection
|
||||
if err := s.DB.First(&p, "brain_instance_ref = ?", instance).Error; err != nil {
|
||||
return ProjectionView{}, err
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
if s.Clock != nil {
|
||||
now = s.Clock().UTC()
|
||||
}
|
||||
stale := s.StaleAfter
|
||||
if stale == 0 {
|
||||
stale = 90 * time.Second
|
||||
}
|
||||
return ProjectionView{Projection: p, Offline: now.Sub(p.ObservedAt) > stale, Stale: now.Sub(p.ObservedAt) > stale}, nil
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
package version
|
||||
|
||||
import (
|
||||
"runtime"
|
||||
|
||||
"gorm.io/gorm"
|
||||
|
||||
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/brain_control"
|
||||
"git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration"
|
||||
common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
|
||||
)
|
||||
|
||||
func init() {
|
||||
_, fileName, _, _ := runtime.Caller(0)
|
||||
migration.Migrate.SetVersion(migration.GetFilename(fileName), migrateBrainRuntime)
|
||||
}
|
||||
|
||||
func migrateBrainRuntime(db *gorm.DB, version string) error {
|
||||
return db.Transaction(func(tx *gorm.DB) error {
|
||||
if err := tx.AutoMigrate(&brain_control.ReplayToken{}, &brain_control.RuntimeProjection{}, &brain_control.SourceRevision{}); err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.Create(&common.Migration{Version: version}).Error
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
package version
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/brain_control"
|
||||
common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
|
||||
"gorm.io/driver/sqlite"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func TestMigrateBrainRuntime(t *testing.T) {
|
||||
db, err := gorm.Open(sqlite.Open("file:brain-runtime-migration?mode=memory&cache=shared"), &gorm.Config{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err = db.AutoMigrate(&common.Migration{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err = migrateBrainRuntime(db, "2026083112000"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, model := range []any{&brain_control.ReplayToken{}, &brain_control.RuntimeProjection{}, &brain_control.SourceRevision{}} {
|
||||
if !db.Migrator().HasTable(model) {
|
||||
t.Fatalf("missing table for %T", model)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user