From 17e31257fd73626f0fb0ab83ada692cf6ad64492 Mon Sep 17 00:00:00 2001 From: Igor Date: Tue, 13 Oct 2026 12:44:27 -0500 Subject: [PATCH] config: poll the chain every second and batch stream frames by time Two defaults that cost latency and dropped frames on mainnet. SORTITION_POLL_INTERVAL 4s -> 1s. Both sortition watchers polled every 4 s against a 2 s block, so a session request or job waited up to two blocks before the worker saw it. STREAM_CHUNK_TOKENS 8 -> 32 and STREAM_CHUNK_INTERVAL 250ms -> 100ms. A frame every 8 tokens is 29 frames/s at 232 tok/s, above the relay's refill rate of 1000 messages/minute (~16.7/s) per consumer, past which it drops frames. With the interval as the binding limit the worker sends about 10 frames/s at any rate up to ~320 tok/s. Co-Authored-By: Claude Opus 5.5 --- internal/config/config.go | 14 +++++++++----- internal/config/config_test.go | 6 +++--- internal/pipeline/stream.go | 4 ++-- internal/pipeline/stream_test.go | 24 ++++++++++++++++++++++++ 4 files changed, 38 insertions(+), 10 deletions(-) diff --git a/internal/config/config.go b/internal/config/config.go index dc2f1e1..0e75f4b 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -120,8 +120,10 @@ type Config struct { // Token-stream coalescing. A chunk frame is published when either // StreamChunkTokens deltas or StreamChunkInterval have accumulated, // whichever comes first. Batching matters because the relay enforces - // a per-consumer rate limit (~1000 messages/minute by default); at - // ~10 tok/s the defaults below emit roughly 1-4 frames/second. + // a per-consumer rate limit (1000 messages/minute by default, refilled + // at ~16.7/s) and drops the frames past it. At the defaults below the + // interval is the binding limit: about 10 frames/second at any + // generation rate up to ~320 tok/s, where the token cap takes over. StreamChunkTokens int StreamChunkInterval time.Duration @@ -482,8 +484,8 @@ func Load() (*Config, error) { cfg.OllamaNumCtx = parseInt("OLLAMA_NUM_CTX", 0, &errs) cfg.OllamaThink = parseBool("OLLAMA_THINK", false, &errs) cfg.OllamaReasoningNumPredict = parseInt("OLLAMA_REASONING_NUM_PREDICT", 4096, &errs) - cfg.StreamChunkTokens = parseInt("STREAM_CHUNK_TOKENS", 8, &errs) - cfg.StreamChunkInterval = parseDuration("STREAM_CHUNK_INTERVAL", "250ms", &errs) + cfg.StreamChunkTokens = parseInt("STREAM_CHUNK_TOKENS", 32, &errs) + cfg.StreamChunkInterval = parseDuration("STREAM_CHUNK_INTERVAL", "100ms", &errs) cfg.StreamReasoning = parseBool("STREAM_REASONING", true, &errs) // Per-model generation-option overrides (heat-tier enforcement). Parsed @@ -556,7 +558,9 @@ func Load() (*Config, error) { } cfg.SortitionStateDir = envOrDefault("SORTITION_STATE_DIR", "data/sortition-state") cfg.SortitionChunkSize = parseUint64("SORTITION_CHUNK_SIZE", 5000, &errs) - cfg.SortitionPollInterval = parseDuration("SORTITION_POLL_INTERVAL", "4s", &errs) + // Below the 2 s block time, so a session request or job is seen within + // about a second of its block instead of up to two blocks later. + cfg.SortitionPollInterval = parseDuration("SORTITION_POLL_INTERVAL", "1s", &errs) cfg.SortitionConfirmations = parseUint64("SORTITION_CONFIRMATIONS", 0, &errs) cfg.SortitionSessionLookbackBlocks = parseUint64("SORTITION_SESSION_LOOKBACK_BLOCKS", 2000, &errs) cfg.SortitionSessionRetryLimit = parseInt("SORTITION_SESSION_RETRY_LIMIT", 10, &errs) diff --git a/internal/config/config_test.go b/internal/config/config_test.go index e6b59b1..f2a5d9a 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -97,8 +97,8 @@ func TestLoad_ValidConfig(t *testing.T) { assert.Equal(t, 1024, cfg.OllamaNumPredict) assert.Equal(t, 0, cfg.OllamaNumCtx, "0 means leave the model's context window alone") assert.False(t, cfg.OllamaThink, "thinking must default to off: the worker never transmits it") - assert.Equal(t, 8, cfg.StreamChunkTokens) - assert.Equal(t, 250*time.Millisecond, cfg.StreamChunkInterval) + assert.Equal(t, 32, cfg.StreamChunkTokens) + assert.Equal(t, 100*time.Millisecond, cfg.StreamChunkInterval) } func TestLoad_StreamingOverrides(t *testing.T) { @@ -690,7 +690,7 @@ func TestConfig_SortitionDefaults(t *testing.T) { require.Equal(t, common.Address{}, cfg.SessionManagerAddress) require.Equal(t, "data/sortition-state", cfg.SortitionStateDir) require.Equal(t, uint64(5000), cfg.SortitionChunkSize) - require.Equal(t, 4*time.Second, cfg.SortitionPollInterval) + require.Equal(t, time.Second, cfg.SortitionPollInterval) require.Equal(t, uint64(0), cfg.SortitionConfirmations) require.Equal(t, 10, cfg.SortitionSessionRetryLimit) } diff --git a/internal/pipeline/stream.go b/internal/pipeline/stream.go index 01c818c..ce0e1ed 100644 --- a/internal/pipeline/stream.go +++ b/internal/pipeline/stream.go @@ -68,8 +68,8 @@ func newTokenCoalescer(maxTokens int, maxInterval time.Duration, emit func(batch // They exist here so a JobHandler built without stream config (tests, the // CLI) still batches sensibly. const ( - defaultStreamChunkTokens = 8 - defaultStreamChunkInterval = 250 * time.Millisecond + defaultStreamChunkTokens = 32 + defaultStreamChunkInterval = 100 * time.Millisecond ) // Add appends a delta and emits the batch if either threshold is now met. diff --git a/internal/pipeline/stream_test.go b/internal/pipeline/stream_test.go index 9d6aff9..e208310 100644 --- a/internal/pipeline/stream_test.go +++ b/internal/pipeline/stream_test.go @@ -155,6 +155,30 @@ func TestNewTokenCoalescer_NonPositiveConfigFallsBackToDefaults(t *testing.T) { assert.Equal(t, defaultStreamChunkInterval, c.maxInterval) } +// The relay refills a consumer's message bucket at 1000/minute (~16.7/s) and +// drops frames past it, so the default batching has to stay under that at any +// generation rate a worker actually reaches. +func TestTokenCoalescer_DefaultsStayUnderRelayRefillRate(t *testing.T) { + t.Parallel() + + const relayRefillPerSecond = 1000.0 / 60.0 + const seconds = 20 + + for _, tokensPerSecond := range []int{10, 34, 150, 232, 320} { + c, clock, batches := newTestCoalescer(defaultStreamChunkTokens, defaultStreamChunkInterval) + gap := time.Second / time.Duration(tokensPerSecond) + for i := 0; i < tokensPerSecond*seconds; i++ { + clock.advance(gap) + require.NoError(t, c.Add("x")) + } + framesPerSecond := float64(len(*batches)) / seconds + assert.LessOrEqual(t, framesPerSecond, 10.5, + "%d tok/s: frame rate must stay near the 100ms interval", tokensPerSecond) + assert.Less(t, framesPerSecond, relayRefillPerSecond, + "%d tok/s: frame rate must stay under the relay's refill rate", tokensPerSecond) + } +} + // --- multi-kind streaming --- // capturingPublisher records every chunk frame the streamer emits, in order,