mirror of
https://github.com/agent-substrate/substrate.git
synced 2026-10-02 03:24:42 +08:00
Make actor template resync interval configurable (#1522)
Actor template discovery currently uses a fixed 20-second resync interval. This PR adds `--template-resync-interval` so deployments can tune that delay, preserving the `20s` default and rejecting nonpositive values. The setting also controls the reconciler's existing fallback retry delay. This is intentionally a small change to start a discussion about how template builds should be dispatched as the template catalog grows. The current resync fetches and decodes every template, including completed ones. At 100,000 templates, a 20-second interval implies roughly 5,000 template rows read per second per replica, assuming scans finish quickly. This is an estimate from the code, not a benchmark. Possible follow-ups: - **Immediate enqueue:** start work after creation, retaining a slower recovery scan for crashes between persistence and enqueue. - **Outbox/watch:** consume changes instead of scanning the catalog. The existing worker outbox still polls every 50 ms and broadcasts events to each subscriber; write overhead and recovery scans need consideration. - **Durable pending-build queue:** atomically record work, claim due jobs using short `FOR UPDATE SKIP LOCKED` transactions, and recover expired leases. This still polls, but queries pending work rather than the full catalog. `LISTEN/NOTIFY` is not the proposed default because it serializes notifying commits. - **Imperative build / long-running operation:** give callers explicit build control or a progress/completion handle. Either still needs reliable execution underneath. The main question is whether template builds need a broadcast change feed or a queue where replicas claim different jobs. Neither alternative is implemented here. [Full research: database costs, execution options, recovery requirements, and sources](https://gist.github.com/EItanya/0a1d6893ede34f2f0e9d9d1929ecb823). Validation: all Go race tests and repository verifiers passed using module mode with `NO_COLOR` unset. CLI checks confirmed the default and rejection of zero/negative intervals; the existing reconciliation test checks a custom interval. Fixes #<issue_number_goes_here> > It's a good idea to open an issue first for discussion. - [ ] Tests pass - [ ] Appropriate changes to documentation are included in the PR --------- Signed-off-by: Eitan Yarmush <eitan.yarmush@solo.io>
This commit is contained in:
@@ -33,8 +33,7 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
templateResyncInterval = 20 * time.Second
|
||||
templateListPageSize = 100
|
||||
templateListPageSize = 100
|
||||
|
||||
// templateWorkerCount is the number of goroutines draining the work
|
||||
// queue.
|
||||
@@ -74,15 +73,17 @@ type goldenActorControl interface {
|
||||
// ActorTemplateReconciler drives stored ActorTemplates through the golden
|
||||
// actor state machine.
|
||||
type ActorTemplateReconciler struct {
|
||||
persistence templateReconcilerStore
|
||||
control goldenActorControl
|
||||
queue workqueue.TypedRateLimitingInterface[resources.ActorTemplateRef]
|
||||
persistence templateReconcilerStore
|
||||
control goldenActorControl
|
||||
queue workqueue.TypedRateLimitingInterface[resources.ActorTemplateRef]
|
||||
resyncInterval time.Duration
|
||||
}
|
||||
|
||||
func NewActorTemplateReconciler(persistence templateReconcilerStore, control goldenActorControl) *ActorTemplateReconciler {
|
||||
func NewActorTemplateReconciler(persistence templateReconcilerStore, control goldenActorControl, resyncInterval time.Duration) *ActorTemplateReconciler {
|
||||
return &ActorTemplateReconciler{
|
||||
persistence: persistence,
|
||||
control: control,
|
||||
persistence: persistence,
|
||||
control: control,
|
||||
resyncInterval: resyncInterval,
|
||||
// Create rate-limiting queue with exponential backoff
|
||||
queue: workqueue.NewTypedRateLimitingQueue(workqueue.DefaultTypedControllerRateLimiter[resources.ActorTemplateRef]()),
|
||||
}
|
||||
@@ -96,7 +97,7 @@ func (r *ActorTemplateReconciler) Start(ctx context.Context) {
|
||||
for range templateWorkerCount {
|
||||
go wait.UntilWithContext(ctx, r.runWorker, time.Second)
|
||||
}
|
||||
wait.UntilWithContext(ctx, r.resync, templateResyncInterval)
|
||||
wait.UntilWithContext(ctx, r.resync, r.resyncInterval)
|
||||
}()
|
||||
}
|
||||
|
||||
@@ -278,7 +279,7 @@ func (r *ActorTemplateReconciler) reconcileOne(ctx context.Context, ref resource
|
||||
return 0, r.fail(ctx, tmpl, reasonUnexpectedState, fmt.Sprintf("golden actor in unexpected state %v", state))
|
||||
|
||||
default:
|
||||
return templateResyncInterval, nil
|
||||
return r.resyncInterval, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -299,7 +299,7 @@ func withFailed(reason string) func(*ateapipb.ActorTemplate) {
|
||||
}
|
||||
|
||||
func newTestTemplateReconciler(persistence templateReconcilerStore, control goldenActorControl) *ActorTemplateReconciler {
|
||||
return NewActorTemplateReconciler(persistence, control)
|
||||
return NewActorTemplateReconciler(persistence, control, 7*time.Second)
|
||||
}
|
||||
|
||||
func TestGoldenSnapshotWarmupFor(t *testing.T) {
|
||||
@@ -467,8 +467,8 @@ func TestReconcileOne(t *testing.T) {
|
||||
name: "unspecified actor state checks back later",
|
||||
template: testTemplate(),
|
||||
control: &fakeGoldenControl{exists: true, goldenState: ateapipb.ActorState_ACTOR_STATE_UNSPECIFIED},
|
||||
wantRequeueMin: templateResyncInterval,
|
||||
wantRequeueMax: templateResyncInterval,
|
||||
wantRequeueMin: 7 * time.Second,
|
||||
wantRequeueMax: 7 * time.Second,
|
||||
},
|
||||
{
|
||||
name: "lease conflict yields without error",
|
||||
|
||||
+8
-1
@@ -64,6 +64,8 @@ import (
|
||||
// maxRPCDeadline is the max deadline for all RPC methods exposed by this server.
|
||||
const maxRPCDeadline = 10 * time.Minute
|
||||
|
||||
const minResyncInterval = 250 * time.Millisecond
|
||||
|
||||
var (
|
||||
listenAddr = pflag.String("grpc-listen-addr", ":443", "Address and port the gRPC server should listen on.")
|
||||
metricsListenAddr = pflag.String("metrics-listen-addr", ":9090", "Address and port the prometheus metrics server should listen on.")
|
||||
@@ -83,6 +85,8 @@ var (
|
||||
drainDelay = pflag.Duration("drain-delay", 13*time.Second, "How long to keep accepting new work after SIGTERM, before starting the gRPC drain.")
|
||||
drainTimeout = pflag.Duration("drain-timeout", 15*time.Second, "Deadline for the graceful gRPC drain on shutdown. In-flight RPCs still running past it are forcefully cancelled.")
|
||||
|
||||
templateResyncInterval = pflag.Duration("template-resync-interval", 20*time.Second, fmt.Sprintf("Interval between actor template resyncs. Must be at least %s.", minResyncInterval))
|
||||
|
||||
showVersion = pflag.Bool("version", false, "Print version and exit.")
|
||||
logLevelFlag = pflag.String("log-level", "info", "Minimum log level: debug, info, warn, or error.")
|
||||
)
|
||||
@@ -99,6 +103,9 @@ func main() {
|
||||
serverboot.Fatal(ctx, "Invalid --log-level", err)
|
||||
}
|
||||
slog.InfoContext(ctx, "ateapi starting", slog.String("version", version.Version))
|
||||
if *templateResyncInterval < minResyncInterval {
|
||||
serverboot.Fatal(ctx, "Invalid --template-resync-interval", fmt.Errorf("must be at least %s", minResyncInterval))
|
||||
}
|
||||
|
||||
// Kept separate from ctx so that in-progress work (clients, informers) is
|
||||
// not cancelled the moment SIGTERM arrives. The drainOnShutdown
|
||||
@@ -226,7 +233,7 @@ func main() {
|
||||
)
|
||||
|
||||
// Drive stored ActorTemplates through the golden actor flow.
|
||||
templateReconciler := controlapi.NewActorTemplateReconciler(persistence, controlSrv)
|
||||
templateReconciler := controlapi.NewActorTemplateReconciler(persistence, controlSrv, *templateResyncInterval)
|
||||
templateReconciler.Start(shutdownCtx)
|
||||
|
||||
lisCfg := &net.ListenConfig{}
|
||||
|
||||
Reference in New Issue
Block a user