Method watchWorkspaceAgentMetadata

coderd/workspaceagents.go:1706–1915

(
	rw http.ResponseWriter,
	r *http.Request,
	connect httpapi.EventSender,
)

}

func (api *API) watchWorkspaceAgentMetadata(
	rw http.ResponseWriter,
	r *http.Request,
	connect httpapi.EventSender,
) {
	// Allow us to interrupt watch via cancel.
	ctx, cancel := context.WithCancel(r.Context())
	defer cancel()
	r = r.WithContext(ctx) // Rewire context for SSE cancellation.

	waws := httpmw.WorkspaceAgentAndWorkspaceParam(r)
	agentIDEncoded := make([]byte, metadatabatcher.UUIDBase64Size)
	err := metadatabatcher.EncodeAgentID(waws.WorkspaceAgent.ID, agentIDEncoded)
	if err != nil {
		httpapi.InternalServerError(rw, err)
		return
	}
	log := api.Logger.Named("workspace_metadata_watcher").With(
		slog.F("workspace_agent_id", waws.WorkspaceAgent.ID),
		slog.F("workspace_id", waws.WorkspaceTable.ID),
	)

	// Send metadata on updates, we must ensure subscription before sending
	// initial metadata to guarantee that events in-between are not missed.
	// The channel carries no data - it's just a signal to fetch all metadata.
	update := make(chan struct{}, 1)

	// Subscribe to the global batched metadata channel.
	// The batcher publishes only to this channel to achieve O(1) NOTIFY scaling.
	cancelBatchSub, err := api.Pubsub.Subscribe(metadatabatcher.MetadataBatchPubsubChannel, func(_ context.Context, byt []byte) {
		if ctx.Err() != nil {
			return
		}

		if len(byt)%metadatabatcher.UUIDBase64Size != 0 {
			log.Error(ctx, "invalid batched pubsub message, pubsub message length was not a multiple of encoded agent UUID length", slog.Error(err))
			return
		}

		// Compare each encoded agentID to our encoded agent ID.
		for i := 0; i < len(byt); i += metadatabatcher.UUIDBase64Size {
			if !bytes.Equal(byt[i:i+metadatabatcher.UUIDBase64Size], agentIDEncoded) {
				continue
			}

			log.Debug(ctx, "received metadata update from batch channel",
				slog.F("agent_id", waws.WorkspaceAgent.ID),
				slog.F("batch_size", len(byt)/metadatabatcher.UUIDBase64Size),
			)

			// Signal to re-fetch all metadata for this agent.
			// Batch notifications don't include which keys changed, so we
			// always fetch all keys for this agent.
			// Attempt to read from the channel first so that we do not block on the write.
			select {
			case <-update:
			default:
			}