2e44147a71
- services/attest-reconciler/ as its own go.mod (admin-api internal/ is not importable; shared code gets extracted when a 3rd consumer appears). - Scheduler accepts "@every <dur>" or a bare Go duration; context-cancellation stops the loop cleanly. - Store client defines PendingRateStore + Postgres skeleton for SELECT ... WHERE rate_change_attestation_ref IS NULL. - Orchestrator + HTTPSigner + tlog.HTTPSubmitter mirror admin-api byte-for-byte so the TLog idempotency key collides on resubmits (kein Doppelsignieren). - Prometheus text-format /metrics endpoint, dependency-free: attest_reconciler_dirty_rows_total, _reconciliation_attempts_total, _reconciliation_failures_total, _idempotent_noop_total. - Tests with fake store + fake signer + fake submitter cover: happy path, already-attested rows are NOT re-signed, TLog idempotency across ticks, TLog-down failure path, pre-cancelled ctx. - Top-level README reflects the new service; CI auto-discovers services/*/go.mod so no workflow change needed. Co-Authored-By: Paperclip <noreply@paperclip.ing>
95 lines
2.6 KiB
Go
95 lines
2.6 KiB
Go
package reconcile
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"log/slog"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
// Scheduler drives Reconciler.RunOnce on a fixed cadence until ctx is
|
|
// cancelled. Spec accepts either "@every <duration>" (robfig/cron compat) or a
|
|
// bare Go duration string ("30s", "5m"). Keeping it to those two shapes buys
|
|
// us zero external deps while still matching the cron.Spec convention the
|
|
// parent issue asked for.
|
|
type Scheduler struct {
|
|
Spec string
|
|
Reconciler *Reconciler
|
|
Logger *slog.Logger
|
|
}
|
|
|
|
// Run blocks until ctx is cancelled. Returns ctx.Err() in that case; returns
|
|
// any other error immediately. Each tick's error is logged and swallowed —
|
|
// the whole point of the reconciler is retry-until-success.
|
|
func (s *Scheduler) Run(ctx context.Context) error {
|
|
if s.Reconciler == nil {
|
|
return errors.New("reconcile.Scheduler: Reconciler is nil")
|
|
}
|
|
interval, err := ParseSpec(s.Spec)
|
|
if err != nil {
|
|
return fmt.Errorf("parse spec %q: %w", s.Spec, err)
|
|
}
|
|
if s.Logger != nil {
|
|
s.Logger.Info("reconciler scheduler started", "interval", interval.String())
|
|
}
|
|
// Fire once immediately so pending rows don't sit a full interval on boot.
|
|
s.tick(ctx)
|
|
t := time.NewTicker(interval)
|
|
defer t.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
if s.Logger != nil {
|
|
s.Logger.Info("reconciler scheduler stopping", "reason", ctx.Err())
|
|
}
|
|
return ctx.Err()
|
|
case <-t.C:
|
|
s.tick(ctx)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Scheduler) tick(ctx context.Context) {
|
|
rep, err := s.Reconciler.RunOnce(ctx)
|
|
if err != nil {
|
|
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
|
|
return
|
|
}
|
|
if s.Logger != nil {
|
|
s.Logger.Error("reconciler tick failed", "err", err)
|
|
}
|
|
return
|
|
}
|
|
if s.Logger != nil && (rep.DirtyRows > 0 || rep.Failures > 0) {
|
|
s.Logger.Info("reconciler tick",
|
|
"dirty_rows", rep.DirtyRows,
|
|
"attempts", rep.Attempts,
|
|
"successes", rep.Successes,
|
|
"failures", rep.Failures)
|
|
}
|
|
}
|
|
|
|
// ParseSpec accepts "@every <duration>" or a bare Go duration. Anything else
|
|
// is rejected so operators who try a full crontab line get a loud error
|
|
// rather than a silent default.
|
|
func ParseSpec(spec string) (time.Duration, error) {
|
|
s := strings.TrimSpace(spec)
|
|
if s == "" {
|
|
return 0, errors.New("empty spec")
|
|
}
|
|
if strings.HasPrefix(s, "@every ") {
|
|
s = strings.TrimSpace(strings.TrimPrefix(s, "@every "))
|
|
} else if strings.ContainsAny(s, "* ") {
|
|
return 0, fmt.Errorf("spec %q: only @every <duration> and bare Go durations are supported", spec)
|
|
}
|
|
d, err := time.ParseDuration(s)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
if d <= 0 {
|
|
return 0, fmt.Errorf("spec %q: duration must be positive", spec)
|
|
}
|
|
return d, nil
|
|
}
|