config: poll the chain every second and batch stream frames by time

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)

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)
 }

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.

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,