fix(din): clock-skew partition clamp + WriteTimeout (telematics-scale review)

cmd/din/main.go

@@ -42,6 +42,18 @@ import (
 // publishAckTimeout bounds the wait for each JetStream publish ack (see jetstream.New).
 const publishAckTimeout = 10 * time.Second
 
+// writeTimeoutMargin is the headroom added to publishAckTimeout for the ingest servers'
+// response WriteTimeout, so a handler that waited the full ack budget still has time to
+// write its 503+Retry-After before the socket deadline fires (scale review #5).
+const writeTimeoutMargin = 5 * time.Second
+
+// attestationPrePublishBudget covers the work the attestation handler does BEFORE the
+// publish ack — an on-chain ERC-1271 signature verify (erc1271CallTimeout, 5s) plus an
+// optional blob PUT — so the attestation WriteTimeout must budget it ON TOP of
+// publishAckTimeout, or a compound-slow attestation has its 503+Retry-After cut off mid-
+// write exactly like the mismatch #5 fixes for the connection path.
+const attestationPrePublishBudget = 15 * time.Second
+
 func main() {
 	log := zerolog.New(os.Stdout).With().Timestamp().Str("app", "din").Logger()
 	// Stamp the build commit on every log line so a running pod reports its version
@@ -345,7 +357,11 @@ func run(log zerolog.Logger) error {
 		MaxBodyBytes:   settings.MaxBodyBytes,
 		RateLimitRPS:   settings.RateLimitRPS,
 		RateLimitBurst: settings.RateLimitBurst,
-		Logger:         log,
+		// Write budget must exceed the publish-ack budget: the handler blocks up to
+		// publishAckTimeout awaiting the JetStream ack before writing a 503+Retry-After,
+		// which the socket WriteTimeout must not cut short (scale review #5).
+		WriteTimeout: publishAckTimeout + writeTimeoutMargin,
+		Logger:       log,
 	}, postOnly(handlers.Connection()))
 	if err != nil {
 		return err
@@ -357,7 +373,9 @@ func run(log zerolog.Logger) error {
 		MaxBodyBytes:           settings.MaxBodyBytes,
 		RateLimitRPS:           settings.RateLimitRPS,
 		RateLimitBurst:         settings.RateLimitBurst,
-		Logger:                 log,
+		// Budget the pre-publish ERC-1271 verify + blob PUT on top of the ack budget (#5).
+		WriteTimeout: publishAckTimeout + attestationPrePublishBudget + writeTimeoutMargin,
+		Logger:       log,
 	}, postOnly(handlers.Attestation()))
 	if err != nil {
 		return err

internal/lake/review_fixes_test.go

@@ -17,7 +17,7 @@ import (
 func TestRowArgs_TruncatesTimeToMillis(t *testing.T) {
 	t.Parallel()
 	ev := testEvent("e1", "dimo.status", "did:1", time.Date(2026, 6, 8, 10, 0, 0, 123456789, time.UTC))
-	args, err := rowArgs(&ev)
+	args, err := rowArgs(&ev, time.Now().UTC())
 	require.NoError(t, err)
 	got, ok := args[1].(time.Time) // column order: subject, "time", ...
 	require.True(t, ok, "second column must be the timestamp")

internal/lake/row.go

@@ -31,9 +31,53 @@ const rawEventColumnCount = 13
 // backfilled bundle must carry an identical value or reader dedup — keyed on
 // (subject, "time", ...) — would treat them as distinct and fail to collapse
 // the native/backfill overlap (SR review #6).
-func rowArgs(event *cloudevent.StoredEvent) ([]driver.Value, error) {
+const (
+	// defaultMaxPastWindow / defaultMaxFutureWindow bound how far from now() the STORED
+	// raw_events "time" (the day("time") partition source) may be before it is treated as a
+	// broken-clock artifact and clamped to now. They are DELIBERATELY WIDE so legit timings
+	// are never touched: offline-buffered readings are commonly days-to-weeks old (past window
+	// = 365d, ~the decoded-retention horizon), and normal skew / dis-parity near-future is
+	// minutes-to-hours (future window = 24h). Only clearly-garbage clocks (unset RTC → 1970,
+	// GPS-week rollover → 1980/2019, factory default → 2000, runaway → 2099) fall outside.
+	//
+	// Crucially this clamp is applied ONLY to the value WRITTEN to raw_events (the partition
+	// key), NOT to the CloudEvent header time. The header time still drives the NATS MsgID and
+	// the decodestream dedup id, so retry dedup and the vehicle-triggers de-dup are unchanged —
+	// clamping the stored time can't make a retry's MsgID drift or collapse distinct siblings.
+	defaultMaxPastWindow   = 365 * 24 * time.Hour
+	defaultMaxFutureWindow = 24 * time.Hour
+)
+
+var (
+	maxPastWindow   = defaultMaxPastWindow
+	maxFutureWindow = defaultMaxFutureWindow
+)
+
+// SetPartitionSafeTimeWindows overrides the broken-clock clamp bounds from validated config.
+// Non-positive values keep the default. Call once at boot before ingest starts.
+func SetPartitionSafeTimeWindows(maxPast, maxFuture time.Duration) {
+	if maxPast > 0 {
+		maxPastWindow = maxPast
+	}
+	if maxFuture > 0 {
+		maxFutureWindow = maxFuture
+	}
+}
+
+// partitionSafeTime returns t if it is within [now-maxPastWindow, now+maxFutureWindow], else
+// now — so a broken device clock can't mint a permanent singleton day-partition in
+// raw_events (PARTITIONED BY day("time")) that maintenance can never merge. Deterministic
+// per (t, now); only the STORED partition value is affected (see the const doc).
+func partitionSafeTime(t, now time.Time) time.Time {
+	if t.Before(now.Add(-maxPastWindow)) || t.After(now.Add(maxFutureWindow)) {
+		return now
+	}
+	return t
+}
+
+func rowArgs(event *cloudevent.StoredEvent, now time.Time) ([]driver.Value, error) {
 	args := make([]driver.Value, rawEventColumnCount)
-	if err := fillRowArgs(args, event); err != nil {
+	if err := fillRowArgs(args, event, now); err != nil {
 		return nil, err
 	}
 	return args, nil
@@ -45,7 +89,7 @@ func rowArgs(event *cloudevent.StoredEvent) ([]driver.Value, error) {
 // columns (data, extras) are passed as []byte to skip a string copy of the
 // largest column — DuckDB's appender validates the UTF-8 VARCHAR contract on the
 // C side whether given a string or []byte, so poison-row detection is unchanged.
-func fillRowArgs(dst []driver.Value, event *cloudevent.StoredEvent) error {
+func fillRowArgs(dst []driver.Value, event *cloudevent.StoredEvent, now time.Time) error {
 	var extrasJSON driver.Value = emptyExtrasJSON
 	if extras := cloudevent.AddNonColumnFieldsToExtras(&event.CloudEventHeader); extras != nil {
 		b, err := json.Marshal(extras)
@@ -71,7 +115,9 @@ func fillRowArgs(dst []driver.Value, event *cloudevent.StoredEvent) error {
 	}
 
 	dst[0] = event.Subject
-	dst[1] = event.Time.UTC().Truncate(time.Millisecond)
+	// Clamp only the STORED (partition-key) time — never the header time, which already drove
+	// the MsgID/dedup upstream (see partitionSafeTime's doc).
+	dst[1] = partitionSafeTime(event.Time, now).UTC().Truncate(time.Millisecond)
 	dst[2] = event.Type
 	dst[3] = event.ID
 	dst[4] = event.Source

internal/lake/row_clamp_test.go

@@ -0,0 +1,61 @@
+package lake
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestPartitionSafeTime(t *testing.T) {
+	t.Parallel()
+	now := time.Date(2026, 7, 9, 12, 0, 0, 0, time.UTC)
+	cases := []struct {
+		name    string
+		in      time.Time
+		wantNow bool
+	}{
+		{"epoch-0 (unset RTC)", time.Unix(0, 0).UTC(), true},
+		{"1980 gps-week rollover", time.Date(1980, 1, 6, 0, 0, 0, 0, time.UTC), true},
+		{"2000 factory default", time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC), true},
+		{"2099 runaway", time.Date(2099, 1, 1, 0, 0, 0, 0, time.UTC), true},
+		{"20d offline buffer", now.AddDate(0, 0, -20), false},
+		{"2m clock skew", now.Add(2 * time.Minute), false},
+		{"12h dis-parity near-future", now.Add(12 * time.Hour), false},
+		{"exactly now", now, false},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := partitionSafeTime(tc.in, now)
+			if tc.wantNow {
+				assert.True(t, got.Equal(now), "garbage time must clamp to now, got %s", got)
+			} else {
+				assert.True(t, got.Equal(tc.in), "in-window time must be untouched, got %s", got)
+			}
+		})
+	}
+}
+
+// TestRowArgs_ClampsStoredTimeNotHeader proves scale-review #1: the broken-clock clamp
+// bounds the raw_events partition key but leaves the CloudEvent HEADER time untouched, so
+// the NATS MsgID and decodestream dedup id (which hash the header time) are unaffected and a
+// retry still dedups instead of double-firing vehicle-triggers.
+func TestRowArgs_ClampsStoredTimeNotHeader(t *testing.T) {
+	t.Parallel()
+	now := time.Date(2026, 7, 9, 12, 0, 0, 0, time.UTC)
+	garbage := time.Date(2099, 1, 1, 0, 0, 0, 0, time.UTC)
+
+	ev := testEvent("g1", "dimo.status", "did:1", garbage)
+	args, err := rowArgs(&ev, now)
+	require.NoError(t, err)
+	assert.True(t, args[1].(time.Time).Equal(now), "stored partition time must clamp to now, got %v", args[1])
+	assert.True(t, ev.Time.Equal(garbage), "HEADER time must be untouched (MsgID/dedup identity), got %v", ev.Time)
+
+	legit := now.AddDate(0, 0, -20)
+	ev2 := testEvent("g2", "dimo.status", "did:1", legit)
+	args2, err := rowArgs(&ev2, now)
+	require.NoError(t, err)
+	assert.True(t, args2[1].(time.Time).Equal(legit.Truncate(time.Millisecond)), "legit time stored as-is (ms-truncated), got %v", args2[1])
+	assert.True(t, ev2.Time.Equal(legit))
+}

internal/lake/writer.go

@@ -9,6 +9,7 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/DIMO-Network/cloudevent"
 	duckdb "github.com/duckdb/duckdb-go/v2"
@@ -168,13 +169,17 @@ func appendAll(ctx context.Context, conn *sql.Conn, table string, events []cloud
 		// the call and never retains it, so refilling it per row avoids ~100k slice
 		// allocations on a full bundle.
 		args := make([]driver.Value, rawEventColumnCount)
+		// One now() per bundle (not per row) for the broken-clock partition clamp: a bundle
+		// is written in one appender pass, so a shared reference is exact and avoids ~100k
+		// time.Now() syscalls on a full bundle.
+		now := time.Now().UTC()
 		for i := range events {
 			// fillRowArgs (marshal) and AppendRow (DuckDB rejecting bad
 			// UTF-8/precision) are deterministic per-row rejections: tag them
 			// ErrPoisonRow so the sink isolates/terminates the row instead of
 			// treating it like a transient outage. The flush below (Close) is the
 			// only I/O here and is deliberately left untagged (transient).
-			if err := fillRowArgs(args, &events[i]); err != nil {
+			if err := fillRowArgs(args, &events[i], now); err != nil {
 				_ = appender.CloseWithCancel(ctx)
 				return fmt.Errorf("lake row %d: %w: %w", i, ErrPoisonRow, err)
 			}

internal/server/connection_server_test.go

@@ -300,3 +300,23 @@ func TestRateLimitMiddleware_Disabled(t *testing.T) {
 		require.Equal(t, http.StatusOK, rec.Code, fmt.Sprintf("request %d must pass with limiting disabled", i))
 	}
 }
+
+// publishAckBudgetForTest mirrors cmd/din's publishAckTimeout (10s) — the max a handler
+// can block on a JetStream ack before writing its 503. Kept here so the server package can
+// assert its write budget exceeds it without importing the cmd package.
+const publishAckBudgetForTest = 10 * time.Second
+
+// TestServers_WriteTimeoutExceedsAckBudget pins the fix for the WriteTimeout(5s) <
+// publishAckTimeout(10s) mismatch: the ingest handler can block up to the publish-ack
+// budget before writing a 503+Retry-After, so the socket WriteTimeout must exceed that
+// (and the read budget) or the orderly backpressure response is never writable — the
+// device sees a raw connection reset and retry-storms (scale review #5).
+func TestServers_WriteTimeoutExceedsAckBudget(t *testing.T) {
+	t.Parallel()
+	_, cfg := connectionTestSetup(t)
+	srv, err := NewConnectionServer(cfg, http.NotFoundHandler())
+	require.NoError(t, err)
+	assert.Equal(t, DefaultWriteTimeout, srv.WriteTimeout)
+	assert.Greater(t, srv.WriteTimeout, srv.ReadTimeout, "write budget must exceed the read budget so a slow-publish 503 is writable")
+	assert.Greater(t, srv.WriteTimeout, publishAckBudgetForTest, "write budget must exceed the publish-ack budget")
+}

internal/server/server.go

@@ -18,7 +18,14 @@ const (
 	DefaultAttestationAddr = ":9442"
 	DefaultOpsAddr         = ":8080"
 	DefaultTimeout         = 5 * time.Second
-	DefaultMaxBodyBytes    = 32 << 20 // 32 MiB
+	// DefaultWriteTimeout bounds writing the response. It is DELIBERATELY larger than the
+	// read/handler-in budget: an ingest handler may block up to the JetStream publish-ack
+	// budget (cmd/din publishAckTimeout, 10s) before mapping a lost ack to 503+Retry-After,
+	// so the socket write deadline must exceed that or the orderly backpressure response is
+	// killed mid-write and the device sees a raw connection reset → retry storm (scale
+	// review #5). Set > publishAckTimeout with margin.
+	DefaultWriteTimeout = 15 * time.Second
+	DefaultMaxBodyBytes = 32 << 20 // 32 MiB
 	// DefaultIdleTimeout bounds how long a keep-alive connection may sit idle
 	// between requests; without it an idle (or slow-trickle) connection is held
 	// open indefinitely, tying up a goroutine/fd (slowloris-style exhaustion).
@@ -39,8 +46,11 @@ type ConnectionConfig struct {
 	// ClientCAFiles are PEM files holding the root CAs that client
 	// certificates must chain to (mutual TLS is required).
 	ClientCAFiles []string
-	// Timeout bounds request read/write; defaults to 5s.
+	// Timeout bounds request READ (header+body); defaults to 5s.
 	Timeout time.Duration
+	// WriteTimeout bounds writing the response; defaults to DefaultWriteTimeout. It must
+	// exceed the publish-ack budget so a slow-publish 503+Retry-After is writable (#5).
+	WriteTimeout time.Duration
 	// MaxBodyBytes caps request body size; defaults to 32 MiB.
 	MaxBodyBytes int64
 	// RateLimitRPS is the per-remote sustained request rate; <= 0 disables
@@ -61,8 +71,11 @@ type AttestationConfig struct {
 	// TokenExchangeKeySetURL provides the public keys for JWT signature
 	// validation (JWKS).
 	TokenExchangeKeySetURL string
-	// Timeout bounds request read/write; defaults to 5s.
+	// Timeout bounds request READ (header+body); defaults to 5s.
 	Timeout time.Duration
+	// WriteTimeout bounds writing the response; defaults to DefaultWriteTimeout. It must
+	// exceed the publish-ack budget so a slow-publish 503+Retry-After is writable (#5).
+	WriteTimeout time.Duration
 	// MaxBodyBytes caps request body size; defaults to 32 MiB.
 	MaxBodyBytes int64
 	// RateLimitRPS is the per-remote sustained request rate; <= 0 disables
@@ -99,6 +112,9 @@ func NewConnectionServer(cfg ConnectionConfig, handler http.Handler) (*http.Serv
 	if cfg.Timeout <= 0 {
 		cfg.Timeout = DefaultTimeout
 	}
+	if cfg.WriteTimeout <= 0 {
+		cfg.WriteTimeout = DefaultWriteTimeout
+	}
 	if cfg.MaxBodyBytes <= 0 {
 		cfg.MaxBodyBytes = DefaultMaxBodyBytes
 	}
@@ -138,7 +154,7 @@ func NewConnectionServer(cfg ConnectionConfig, handler http.Handler) (*http.Serv
 		},
 		ReadTimeout:       cfg.Timeout,
 		ReadHeaderTimeout: cfg.Timeout,
-		WriteTimeout:      cfg.Timeout,
+		WriteTimeout:      cfg.WriteTimeout,
 		IdleTimeout:       DefaultIdleTimeout,
 		MaxHeaderBytes:    DefaultMaxHeaderBytes,
 	}, nil
@@ -157,6 +173,9 @@ func NewAttestationServer(cfg AttestationConfig, handler http.Handler) (*http.Se
 	if cfg.Timeout <= 0 {
 		cfg.Timeout = DefaultTimeout
 	}
+	if cfg.WriteTimeout <= 0 {
+		cfg.WriteTimeout = DefaultWriteTimeout
+	}
 	if cfg.MaxBodyBytes <= 0 {
 		cfg.MaxBodyBytes = DefaultMaxBodyBytes
 	}
@@ -186,7 +205,7 @@ func NewAttestationServer(cfg AttestationConfig, handler http.Handler) (*http.Se
 		Handler:           h,
 		ReadTimeout:       cfg.Timeout,
 		ReadHeaderTimeout: cfg.Timeout,
-		WriteTimeout:      cfg.Timeout,
+		WriteTimeout:      cfg.WriteTimeout,
 		IdleTimeout:       DefaultIdleTimeout,
 		MaxHeaderBytes:    DefaultMaxHeaderBytes,
 	}, nil