feat: implement the refactored chatd stream loop by hugodutka · Pull Request #26110 · coder/coder

hugodutka · 2026-06-05T17:51:56Z

PR 4 of the chatd refactor. Implements the /chats/$id/stream and /chats/$id/stream/parts endpoints.

The CI doesn't pass on purpose. Only the tip of the chatd refactor PR stack will pass CI.

Previous PR in the stack: #26109

github-actions · 2026-06-05T17:52:08Z

Docs preview

📖 View docs preview for docs/admin/integrations/prometheus.md

mafredri

I have a couple of files left to review in this PR, but I've looked through most of it.

I will echo sentiment from previous PR review, lacking in documentation, comments and motivation behind choices.

mafredri · 2026-06-09T14:13:03Z

+
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	go httpapi.HeartbeatClose(ctx, e.logger, cancel, conn)


Suggestion: Could use WSWatcher directly instead of this old method.

mafredri · 2026-06-09T14:17:11Z

+
+func (t streamPartsChannelClientTransport) WriteControl(ctx context.Context, control streamPartsControl) error {
+	select {
+	case <-ctxDone(ctx):


It's unusual to design API's that accept nil ctx. Why are we doing it here?

Address now. That's slop

mafredri · 2026-06-09T14:21:11Z

+		select {
+		case <-f.configure:
+		default:
+		}


This is inherently racy anyway, so why do we do the select twice with depeltion in-between vs. depleting first and selecting once?

Address now

mafredri · 2026-06-09T14:22:27Z

+func (f *streamRelayForwarder) Close() {
+	if f == nil {
+		return
+	}


Closing a nil forwarder seems like a bug, and this masks it. Is there any reason to keep these types of checks?

mafredri · 2026-06-09T14:31:42Z

+	events := make(chan codersdk.ChatStreamEvent, 128)
+	logger := p.logger.With(slog.F("chat_id", chatID))
+
+	updateCh := make(chan streamSyncHint, 32)


I'd like to see motivations for the selected buffer values here (and in other places). To the reader 32 seems quite arbitrary considering we don't even drop any updates.

mafredri · 2026-06-09T14:36:14Z

+	}
+	ch := make(chan streamSyncHint)
+	close(ch)
+	return ch, func() {}


This feels weird. Is there any reason to runt he server without a streamSyncPoller? If this is for tets then a fakeStreamSyncPoller should be used instead.

Address now

mafredri · 2026-06-09T14:38:42Z

+	clock := p.clock
+	if clock == nil {
+		clock = quartz.NewReal()
+	}


Masking a bug in server startup?

Address now

mafredri · 2026-06-09T14:40:55Z

+func (p *streamSyncPoller) Start() {
+	if p == nil {
+		return
+	}


This is the last time I'll flag these oddities for this PR, but it's very weird handle these in production code. This smells like production code being adapted for tests rather than restructuring so that tests set up for production.

mafredri · 2026-06-09T14:44:20Z

+		hint := streamSyncHintFromPollRow(row)
+		for _, subscriber := range subscribers[row.ID] {
+			select {
+			case subscriber.hints <- hint:


A reader might want to know why/if it's fine that hints may have stale data and that we're not trying to deplete it first and send an up-to-date hint.

mafredri · 2026-06-09T14:48:37Z

-				return uuid.New()
-			}
-			return id
+			return api.AGPL.ID


Is this change correct?

Address now

johnstcn · 2026-06-09T16:11:33Z

+func (p *Server) syncStreamWithRetry(
+	ctx context.Context,
+	loop *streamLoop,
+	hint streamSyncHint,
+) ([]codersdk.ChatStreamEvent, streamRelayTarget, bool, error) {
+	var (
+		syncEvents []codersdk.ChatStreamEvent
+		target     streamRelayTarget
+		changed    bool
+		err        error
+	)
+	for attempt := 1; attempt <= streamSyncRetryMaxAttempts; attempt++ {
+		//nolint:gocritic // The subscriber was authorized before the loop started; follow-up syncs need chatd-scoped reads for consistency.
+		syncEvents, target, changed, err = loop.sync(dbauthz.AsChatd(ctx), hint)
+		if err == nil || ctx.Err() != nil {
+			return syncEvents, target, changed, err
+		}
+		p.logger.Warn(ctx, "failed to sync chat stream",
+			slog.F("attempt", attempt),
+			slog.Error(err),
+		)
+		if attempt == streamSyncRetryMaxAttempts {
+			break
+		}
+		if !p.waitBeforeStreamSyncRetry(ctx, attempt) {
+			return nil, loop.currentRelayTarget(), false, ctx.Err()
+		}
+	}
+	return nil, loop.currentRelayTarget(), false, err
+}


suggestion: Could we use github.com/coder/retry here instead of the custom backoff loop, with Jitter set? The current syncStreamWithRetry retry schedule is deterministic across subscribers/replicas, so a DB hiccup can synchronize retries. retry.Retrier already supports jitter and context cancellation.

This would not solve the per-subscriber duplicate sync work, but it would reduce thundering-herd behavior during transient DB failures. I’d also add startup jitter to streamSyncPoller, since every chatd instance currently starts a fixed 10s ticker in New().

johnstcn · 2026-06-09T16:19:54Z

+				slog.F("worker_id", target.workerID.UUID),
+				slog.Error(err),
+			)
+			scheduleRetry()


We should check the kind of dialer error - is it something that a retry might fix or not?

Address now

johnstcn · 2026-06-09T16:22:16Z

+	pollerCh, unregisterPoller := p.registerStreamSyncPoller(chatID)
+	loop := newStreamLoop(chat, p.db, logger, afterMessageID)
+	//nolint:gocritic // The HTTP route authorizes the chat before subscribing; the stream loop needs chatd-scoped reads for one consistent snapshot.
+	initial, target, _, err := loop.syncDB(dbauthz.AsChatd(ctx))


Do we need to call loop.syncDB immediately here?

stream_subscribe.go:62-65 creates one streamLoop per subscriber and immediately syncs DB

stream_subscribe.go:93-103 every subscriber processes pubsub/poller hints

stream_subscribe.go:137-173 every subscriber runs syncStreamWithRetry

stream_loop.go:98-112 every subscriber's streamLoop calls syncDB

stream_loop.go:141+ syncDB performs actual DB snapshot reads

Address now

mafredri

I've gone over all the files now, this is my final review on this PR.

mafredri · 2026-06-10T12:51:08Z

+			retryTimer = nil
+			retryC = nil
+		}
+	}


Suggested change

}

}

defer stopRetry()

Address now

mafredri · 2026-06-10T12:51:53Z

+		session = nil
+		sessionParts = nil
+		connected = streamRelayTarget{}
+	}


Suggested change

}

}

defer closeSession()

Or alternatively, defer func() { stopRetry(); closeSession(); }() if you want to change the order of these calls.

This allows simplifying the for loop below and protects against leaks in future code changes.

Address now

mafredri · 2026-06-10T12:54:14Z

+			stopRetry()
+			if !target.needsRelay() {
+				closeSession()
+				continue
+			}


This is already part of connect, why the repetition?

Address now

mafredri · 2026-06-10T12:54:50Z

+			}
+		}
+	}
+}


This function seems fragile and I'd love to see it refactored.

github-actions Bot assigned hugodutka Jun 5, 2026

hugodutka force-pushed the hugodutka/chatd-refactor-4 branch from eebc445 to 38dc20b Compare June 5, 2026 18:05

hugodutka force-pushed the hugodutka/chatd-refactor-3 branch from 0e10473 to 0840442 Compare June 5, 2026 18:05

hugodutka changed the title ~~pr 4~~ feat: implement the refactored chatd stream loop Jun 8, 2026

hugodutka mentioned this pull request Jun 8, 2026

feat: adapt the frontend to the refactored chatd #26111

Merged

hugodutka marked this pull request as ready for review June 8, 2026 12:39

hugodutka requested review from johnstcn and mafredri June 8, 2026 12:39

hugodutka force-pushed the hugodutka/chatd-refactor-3 branch from 0840442 to 965ee76 Compare June 8, 2026 13:07

hugodutka force-pushed the hugodutka/chatd-refactor-4 branch from 38dc20b to 8bd7ca4 Compare June 8, 2026 13:07

hugodutka force-pushed the hugodutka/chatd-refactor-3 branch from 965ee76 to 3bbda07 Compare June 9, 2026 12:03

hugodutka force-pushed the hugodutka/chatd-refactor-4 branch from 8bd7ca4 to 8b3e9f7 Compare June 9, 2026 12:03

mafredri reviewed Jun 9, 2026

View reviewed changes

johnstcn reviewed Jun 9, 2026

View reviewed changes

hugodutka force-pushed the hugodutka/chatd-refactor-3 branch from 3bbda07 to c48520d Compare June 9, 2026 16:32

hugodutka force-pushed the hugodutka/chatd-refactor-4 branch from 8b3e9f7 to 44cbf6c Compare June 9, 2026 16:32

mafredri reviewed Jun 10, 2026

View reviewed changes

hugodutka force-pushed the hugodutka/chatd-refactor-3 branch from c48520d to 2cf1f47 Compare June 11, 2026 11:43

hugodutka force-pushed the hugodutka/chatd-refactor-4 branch from 44cbf6c to 110e90b Compare June 11, 2026 11:43

hugodutka force-pushed the hugodutka/chatd-refactor-3 branch from 2cf1f47 to 47d102f Compare June 11, 2026 12:41

hugodutka force-pushed the hugodutka/chatd-refactor-4 branch from 110e90b to 1c07612 Compare June 11, 2026 12:41

hugodutka force-pushed the hugodutka/chatd-refactor-3 branch from 47d102f to bb37788 Compare June 11, 2026 16:27

hugodutka force-pushed the hugodutka/chatd-refactor-4 branch from 1c07612 to 51648bc Compare June 11, 2026 16:27

hugodutka force-pushed the hugodutka/chatd-refactor-3 branch from bb37788 to fd9d643 Compare June 11, 2026 17:03

hugodutka force-pushed the hugodutka/chatd-refactor-4 branch from 51648bc to f7a987d Compare June 11, 2026 17:03

hugodutka force-pushed the hugodutka/chatd-refactor-3 branch from fd9d643 to c9de302 Compare June 11, 2026 17:04

hugodutka force-pushed the hugodutka/chatd-refactor-4 branch from f7a987d to 4f2ce0e Compare June 11, 2026 17:04

hugodutka mentioned this pull request Jun 11, 2026

fix: chatd refactor #26270

Merged

hugodutka force-pushed the hugodutka/chatd-refactor-3 branch from c9de302 to b688bfb Compare June 11, 2026 17:11

hugodutka force-pushed the hugodutka/chatd-refactor-4 branch from 4f2ce0e to db66ec8 Compare June 11, 2026 17:11

Base automatically changed from hugodutka/chatd-refactor-3 to hugodutka/chatd-refactor June 11, 2026 17:12

pr 4

921fd30

hugodutka force-pushed the hugodutka/chatd-refactor-4 branch from db66ec8 to 921fd30 Compare June 11, 2026 17:13

hugodutka merged commit ef45425 into hugodutka/chatd-refactor Jun 11, 2026
2 of 3 checks passed

hugodutka deleted the hugodutka/chatd-refactor-4 branch June 11, 2026 17:13

github-actions Bot locked and limited conversation to collaborators Jun 11, 2026

Uh oh!

Conversation

hugodutka commented Jun 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

github-actions Bot commented Jun 5, 2026

Docs preview

Uh oh!

mafredri left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mafredri left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hugodutka commented Jun 5, 2026 •

edited

Loading