opc/yovision
2
0
Fork 0
Code Issues 1 Pull Requests 8 Actions Packages Projects Releases Wiki Activity

feat(sense): add reconciliation safety controls [T-012]
Harness governance / validate (push) Has been cancelled
Details
Harness governance / validate (pull_request) Has been cancelled
Details

Browse Source
This commit is contained in:
QiuSW
2026-08-07 23:00:03 +08:00
parent 677ed732f7
commit 12857fdf32
35 changed files with 2817 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Sense/internal/reconcile/reconciler.go
+137 -26
View File
@@ -3,12 +3,15 @@ package reconcile
import (
"context"
"crypto/rand"
"encoding/base64"
"errors"
"fmt"
"net/url"
"time"
"yovision/sense/internal/device"
"yovision/sense/internal/metrics"
"yovision/sense/internal/onvif"
"yovision/sense/internal/store"
)
@@ -16,9 +19,11 @@ import (
const defaultBatchSize = 128
type Repository interface {
ListDueReconcile(ctx context.Context, now time.Time, limit int) ([]store.ReconcileCandidate, error)
MarkReconciled(ctx context.Context, id string, generation int64, now time.Time) error
MarkReconcileFailure(ctx context.Context, id string, failureCount int, nextAttempt time.Time, errorCode string, now time.Time) error
ClaimDueReconcile(context.Context, store.ReconcileClaim) ([]store.ReconcileCandidate, error)
RenewReconcileLease(context.Context, string, string, string, time.Time, time.Duration) (bool, error)
CompleteReconcile(context.Context, string, int64, string, string, time.Time) error
FailReconcile(context.Context, string, int, time.Time, string, string, string, time.Time) error
ConvergenceSnapshot(context.Context) (store.ConvergenceSnapshot, error)
}
type MediaPaths interface {
@@ -27,48 +32,134 @@ type MediaPaths interface {
}
type Reconciler struct {
repository Repository
discovery onvif.Adapter
media MediaPaths
now func() time.Time
baseBackoff time.Duration
maxBackoff time.Duration
batchSize int
repository Repository
discovery onvif.Adapter
media MediaPaths
now func() time.Time
baseBackoff time.Duration
maxBackoff time.Duration
batchSize int
instanceID string
leaseDuration time.Duration
operationTimeout time.Duration
metrics *metrics.Registry
}
type Options struct {
InstanceID string
LeaseDuration time.Duration
OperationTimeout time.Duration
Metrics *metrics.Registry
}
func New(repository Repository, discovery onvif.Adapter, media MediaPaths) *Reconciler {
return NewWithOptions(repository, discovery, media, Options{})
}
func NewWithOptions(
repository Repository,
discovery onvif.Adapter,
media MediaPaths,
options Options,
) *Reconciler {
if options.InstanceID == "" {
options.InstanceID = "single"
}
if options.LeaseDuration <= 0 {
options.LeaseDuration = 30 * time.Second
}
if options.OperationTimeout <= 0 {
options.OperationTimeout = 20 * time.Second
}
if options.OperationTimeout >= options.LeaseDuration {
options.OperationTimeout = options.LeaseDuration / 2
}
return &Reconciler{
repository: repository, discovery: discovery, media: media,
now: time.Now, baseBackoff: time.Second, maxBackoff: time.Minute, batchSize: defaultBatchSize,
instanceID: options.InstanceID, leaseDuration: options.LeaseDuration,
operationTimeout: options.OperationTimeout, metrics: options.Metrics,
}
}
func (r *Reconciler) RunOnce(ctx context.Context) error {
func (r *Reconciler) RunOnce(ctx context.Context) (runErr error) {
started := time.Now()
defer func() {
if r.metrics != nil {
r.metrics.ObserveReconcileRun(runErr, time.Since(started))
}
}()
now := r.now().UTC()
candidates, err := r.repository.ListDueReconcile(ctx, now, r.batchSize)
token, err := newClaimToken()
if err != nil {
return fmt.Errorf("list reconciliation candidates: %w", err)
return err
}
candidates, err := r.repository.ClaimDueReconcile(ctx, store.ReconcileClaim{
Owner: r.instanceID, Token: token, Now: now,
LeaseDuration: r.leaseDuration, Limit: r.batchSize,
})
if err != nil {
return fmt.Errorf("claim reconciliation candidates: %w", err)
}
var runErrors []error
for _, candidate := range candidates {
if err := ctx.Err(); err != nil {
return err
}
if err := r.reconcileOne(ctx, candidate, now); err != nil {
itemNow := r.now().UTC()
renewed, err := r.repository.RenewReconcileLease(
ctx, candidate.Device.ID, r.instanceID, token, itemNow, r.leaseDuration,
)
if err != nil {
r.observeItem("error")
runErrors = append(runErrors, fmt.Errorf("renew reconcile device lease: %w", err))
continue
}
if !renewed {
r.observeItem("lease_lost")
continue
}
itemCtx, cancel := context.WithTimeout(ctx, r.operationTimeout)
err = r.reconcileOne(itemCtx, ctx, candidate, itemNow, r.instanceID, token)
cancel()
if errors.Is(err, store.ErrReconcileLeaseLost) {
r.observeItem("lease_lost")
continue
}
if err != nil {
r.observeItem("error")
runErrors = append(runErrors, fmt.Errorf("reconcile device %s: %w", candidate.Device.ID, err))
} else {
r.observeItem("success")
}
}
if r.metrics != nil {
snapshot, err := r.repository.ConvergenceSnapshot(ctx)
if err != nil {
runErrors = append(runErrors, fmt.Errorf("read convergence metrics: %w", err))
} else {
r.metrics.SetConvergence(snapshot.Total, snapshot.Unconverged)
}
}
return errors.Join(runErrors...)
}
func (r *Reconciler) reconcileOne(ctx context.Context, candidate store.ReconcileCandidate, now time.Time) error {
func (r *Reconciler) reconcileOne(
ctx context.Context,
persistCtx context.Context,
candidate store.ReconcileCandidate,
now time.Time,
owner, token string,
) error {
if candidate.Device.DesiredState == device.DesiredDisabled {
if err := r.media.DeletePath(ctx, candidate.Device.PathName); err == nil {
return r.repository.MarkReconciled(ctx, candidate.Device.ID, candidate.Device.Generation, now)
} else if ctx.Err() != nil {
return ctx.Err()
return r.repository.CompleteReconcile(
persistCtx, candidate.Device.ID, candidate.Device.Generation, owner, token, now,
)
} else if persistCtx.Err() != nil {
return persistCtx.Err()
} else {
return r.persistFailure(ctx, candidate, now, err)
return r.persistFailure(persistCtx, candidate, now, owner, token, err)
}
}
result, err := r.discovery.Probe(ctx, onvif.Target{
@@ -81,16 +172,22 @@ func (r *Reconciler) reconcileOne(ctx context.Context, candidate store.Reconcile
_, err = r.media.EnsurePath(ctx, candidate.Device.PathName, result.StreamURI)
}
if err == nil {
return r.repository.MarkReconciled(ctx, candidate.Device.ID, candidate.Device.Generation, now)
return r.repository.CompleteReconcile(
persistCtx, candidate.Device.ID, candidate.Device.Generation, owner, token, now,
)
}
if ctx.Err() != nil {
return ctx.Err()
if persistCtx.Err() != nil {
return persistCtx.Err()
}
return r.persistFailure(ctx, candidate, now, err)
return r.persistFailure(persistCtx, candidate, now, owner, token, err)
}
func (r *Reconciler) persistFailure(
ctx context.Context, candidate store.ReconcileCandidate, now time.Time, err error,
ctx context.Context,
candidate store.ReconcileCandidate,
now time.Time,
owner, token string,
err error,
) error {
failureCount := candidate.FailureCount + 1
nextAttempt := now.Add(r.backoff(failureCount))
@@ -99,14 +196,28 @@ func (r *Reconciler) persistFailure(
if !errors.As(err, &onvifError) {
errorCode = "media_error"
}
if markErr := r.repository.MarkReconcileFailure(
ctx, candidate.Device.ID, failureCount, nextAttempt, errorCode, now,
if markErr := r.repository.FailReconcile(
ctx, candidate.Device.ID, failureCount, nextAttempt, errorCode, owner, token, now,
); markErr != nil {
return errors.Join(err, fmt.Errorf("persist reconcile failure: %w", markErr))
}
return err
}
func (r *Reconciler) observeItem(result string) {
if r.metrics != nil {
r.metrics.ObserveReconcileItem(result)
}
}
func newClaimToken() (string, error) {
value := make([]byte, 18)
if _, err := rand.Read(value); err != nil {
return "", errors.New("generate reconcile claim token")
}
return base64.RawURLEncoding.EncodeToString(value), nil
}
func validateStreamURI(value string) error {
parsed, err := url.Parse(value)
if err != nil || parsed.Host == "" || (parsed.Scheme != "rtsp" && parsed.Scheme != "rtsps") {
Sense/internal/reconcile/reconciler_test.go
+80
View File
@@ -162,6 +162,86 @@ func TestDisabledDeviceDeletesOnlyItsExactPath(t *testing.T) {
}
}
type lostLeaseRepository struct {
completed bool
failed bool
}
func (r *lostLeaseRepository) ClaimDueReconcile(
context.Context,
store.ReconcileClaim,
) ([]store.ReconcileCandidate, error) {
return []store.ReconcileCandidate{{Device: device.Device{
ID: "camera-lost", DesiredState: device.DesiredEnabled,
EndpointRef: "onvif://camera-lost", PathName: "camera-lost", Generation: 1,
}}}, nil
}
func (r *lostLeaseRepository) RenewReconcileLease(
context.Context, string, string, string, time.Time, time.Duration,
) (bool, error) {
return false, nil
}
func (r *lostLeaseRepository) CompleteReconcile(
context.Context, string, int64, string, string, time.Time,
) error {
r.completed = true
return nil
}
func (r *lostLeaseRepository) FailReconcile(
context.Context, string, int, time.Time, string, string, string, time.Time,
) error {
r.failed = true
return nil
}
func (r *lostLeaseRepository) ConvergenceSnapshot(context.Context) (store.ConvergenceSnapshot, error) {
return store.ConvergenceSnapshot{}, nil
}
func TestLostLeaseSkipsAllExternalAndStoreMutations(t *testing.T) {
t.Parallel()
repository := &lostLeaseRepository{}
media := &recordingMedia{}
reconciler := New(repository, onvif.NewFake(nil), media)
if err := reconciler.RunOnce(context.Background()); err != nil {
t.Fatal(err)
}
if media.calls != 0 || repository.completed || repository.failed {
t.Fatalf("lost lease performed a side effect: media=%d completed=%v failed=%v",
media.calls, repository.completed, repository.failed)
}
}
func TestPerItemTimeoutPersistsRetryWhenParentIsAlive(t *testing.T) {
t.Parallel()
repository := openRepository(t, filepath.Join(t.TempDir(), "sense.db"))
createReconcileDevice(t, repository)
discovery := onvif.NewFake(map[string]onvif.FakeScenario{
"onvif://camera-1": {
DelayMillis: 100,
Result: onvif.ProbeResult{StreamURI: "rtsp://media.invalid/camera-1"},
},
})
reconciler := NewWithOptions(repository, discovery, &recordingMedia{}, Options{
InstanceID: "ins_test", LeaseDuration: time.Second, OperationTimeout: 5 * time.Millisecond,
})
now := time.Date(2026, 8, 7, 0, 0, 0, 0, time.UTC)
reconciler.now = func() time.Time { return now }
if err := reconciler.RunOnce(context.Background()); err == nil {
t.Fatal("expected bounded operation timeout")
}
candidates, err := repository.ListDueReconcile(context.Background(), now.Add(time.Hour), 1)
if err != nil {
t.Fatal(err)
}
if len(candidates) != 1 || candidates[0].FailureCount != 1 {
t.Fatalf("operation timeout did not persist retry state: %+v", candidates)
}
}
func openRepository(t *testing.T, path string) *store.SQLite {
t.Helper()
repository, err := store.OpenSQLite(context.Background(), "file:"+filepath.ToSlash(path))