mirror of
https://github.com/coder/coder.git
synced 2026-09-24 15:04:27 +08:00
fix(coderd): give chat message ids an append-order guarantee (#27495)
Chat message ordering was derived from `created_at`, which is `now()` and therefore the transaction start time. That makes it unusable as an append-order column for two independent reasons: every row in one `InsertChatMessages` batch shares a single timestamp, and two concurrent transactions can commit in the opposite order to the one they started in. This PR gives `chat_messages.id` a real append-order guarantee and moves the history reads onto it. ## Changes **`InsertChatMessages` had no input-order guarantee.** Callers index the returned slice by input position. That only worked because PostgreSQL happens to evaluate the `BIGSERIAL` default in row order. Ids are now allocated up front and the k-th smallest is assigned to input index k, so the pairing does not depend on where the column default is evaluated. Returned rows are explicitly `ORDER BY id`. **Three history reads now order by `id`.** | Query | Was | Now | |---|---|---| | `GetChatMessagesByChatID` | `created_at ASC` | `id ASC` | | `GetChatMessagesByRevisionForStream` | `created_at ASC, id ASC` | `id ASC` | | `GetLastChatMessageByRole` | `created_at DESC, id DESC` | `id DESC` | `GetChatMessagesByChatID` paginated by `id` while ordering by `created_at`, which is incoherent on its own terms. The other two matter because of who consumes them. The stream query supplies incremental updates on the same socket that emits a full `GetChatMessagesByChatID` snapshot on history reset, so once that snapshot moved to `id` the two disagreed under timestamp skew. `GetLastChatMessageByRole` returns an id that is then used as an id cursor, both as `AfterID` when synthesizing tool cancellations and as `chats.last_read_message_id`, where a stale anchor leaves later assistant messages permanently unread. A tie-breaker would not have fixed either one. It only resolves equal timestamps; leading with `created_at` is the actual defect. **`GetLastChatMessageByRole` loses its index, so this adds one.** `ORDER BY created_at DESC, id DESC` could take an ordered scan of `idx_chat_messages_chat_created`. Nothing in the schema can supply `ORDER BY id DESC LIMIT 1` for a given `chat_id` and `role`, so the planner switches to a backward scan of the primary key and filters every newer row in the table, scanning all of it when the chat has no message in that role, which is the routine case for a fresh chat. Migration `000559` adds `(chat_id, role, id DESC) WHERE deleted = false`, the same shape as the existing `idx_chat_messages_user_prompts`. This matters because the query is hot: it runs on every stream connect and disconnect, and once per turn when synthesizing tool cancellations. `GetChatMessagesForPromptByChatID` has the same defect and is fixed in the stacked PR, because its compaction boundary change is semantic and deserves a separate review. Auto-archive stays timestamp-based deliberately: it measures activity, not order. Wrapping the insert in a CTE (needed because `INSERT` cannot take `ORDER BY`) makes sqlc synthesize `InsertChatMessagesRow`. It is structurally identical to `ChatMessage`, so the call sites use a direct struct conversion that stops compiling if the two ever diverge. ## Testing Behavior tests write `created_at` values inverted against id order, so a reader that leads with `created_at` returns the batch backwards. All three queries were verified red by reverting the `ORDER BY` and regenerating: the stream query returned `[3,2,1]` for `[1,2,3]`, and `GetLastChatMessageByRole` picked id 1 instead of id 3. `TestInsertChatMessagesOrderContract` asserts against the generated SQL, covering what a behavior test cannot: PostgreSQL evaluates the id default in row order anyway, so a batch still looks ordered once the guarantee is removed. `TestChatMessagesSequenceCacheIsOne` guards the cross-batch half of the invariant. Ids follow chat row lock order only while the sequence hands out one value at a time; sequence cache blocks are per session, so with a cache above one a backend holding stale cached values can lock second and still commit lower ids. Bumping a sequence cache is an ordinary throughput tweak, and it would silently corrupt history order. The index was checked on a 200k row fixture. Without it, the zero-match lookup filters all 200,000 rows over 2763 buffers; with it, the plan is an index scan with both `chat_id` and `role` in the index condition, no sort node, and 3 buffers. Note that the within-batch mapping does not depend on the cache size. It is established by `ROW_NUMBER() OVER (ORDER BY id)` over the allocated ids, so it holds regardless of `nextval` evaluation order. ## Note on the deleted subagent hand-sort The subagent history reader's hand-sort stays deleted, but calling it redundant was imprecise. It sorted by `created_at` then `id`, so it is only equivalent to `id` ordering when the two agree. When they disagree the old code selected a different "latest assistant". This is a deliberate behavior change to match the new invariant, not dead-code removal. > Opened by Mux on behalf of Mike.
This commit is contained in:
Michael Suchacz
Generated
+109
-54
@@ -8126,7 +8126,7 @@ WHERE
|
||||
AND visibility IN ('user', 'both')
|
||||
AND deleted = false
|
||||
ORDER BY
|
||||
created_at ASC
|
||||
id ASC
|
||||
`
|
||||
|
||||
type GetChatMessagesByChatIDParams struct {
|
||||
@@ -8134,6 +8134,9 @@ type GetChatMessagesByChatIDParams struct {
|
||||
AfterID int64 `db:"after_id" json:"after_id"`
|
||||
}
|
||||
|
||||
// Ordered by id to match the @after_id cursor. created_at is the transaction
|
||||
// start time, so it can disagree with append order when a transaction takes the
|
||||
// chat row lock later than one that started after it.
|
||||
func (q *sqlQuerier) GetChatMessagesByChatID(ctx context.Context, arg GetChatMessagesByChatIDParams) ([]ChatMessage, error) {
|
||||
rows, err := q.db.QueryContext(ctx, getChatMessagesByChatID, arg.ChatID, arg.AfterID)
|
||||
if err != nil {
|
||||
@@ -8345,7 +8348,7 @@ WHERE
|
||||
AND revision > $2::bigint
|
||||
AND visibility IN ('user', 'both')
|
||||
ORDER BY
|
||||
created_at ASC, id ASC
|
||||
id ASC
|
||||
`
|
||||
|
||||
type GetChatMessagesByRevisionForStreamParams struct {
|
||||
@@ -8353,6 +8356,7 @@ type GetChatMessagesByRevisionForStreamParams struct {
|
||||
AfterRevision int64 `db:"after_revision" json:"after_revision"`
|
||||
}
|
||||
|
||||
// Stream deltas and reset snapshots must use the same message order.
|
||||
func (q *sqlQuerier) GetChatMessagesByRevisionForStream(ctx context.Context, arg GetChatMessagesByRevisionForStreamParams) ([]ChatMessage, error) {
|
||||
rows, err := q.db.QueryContext(ctx, getChatMessagesByRevisionForStream, arg.ChatID, arg.AfterRevision)
|
||||
if err != nil {
|
||||
@@ -9867,7 +9871,7 @@ WHERE
|
||||
AND role = $2::chat_message_role
|
||||
AND deleted = false
|
||||
ORDER BY
|
||||
created_at DESC, id DESC
|
||||
id DESC
|
||||
LIMIT
|
||||
1
|
||||
`
|
||||
@@ -9877,6 +9881,8 @@ type GetLastChatMessageByRoleParams struct {
|
||||
Role ChatMessageRole `db:"role" json:"role"`
|
||||
}
|
||||
|
||||
// The returned id becomes both an AfterID cursor and last_read_message_id, so
|
||||
// "last" must use id order.
|
||||
func (q *sqlQuerier) GetLastChatMessageByRole(ctx context.Context, arg GetLastChatMessageByRoleParams) (ChatMessage, error) {
|
||||
row := q.db.QueryRowContext(ctx, getLastChatMessageByRole, arg.ChatID, arg.Role)
|
||||
var i ChatMessage
|
||||
@@ -10375,7 +10381,7 @@ WITH batch AS (
|
||||
SELECT
|
||||
(
|
||||
SELECT val
|
||||
FROM UNNEST($3::uuid[])
|
||||
FROM UNNEST($1::uuid[])
|
||||
WITH ORDINALITY AS t(val, ord)
|
||||
WHERE val != '00000000-0000-0000-0000-000000000000'::uuid
|
||||
ORDER BY ord DESC
|
||||
@@ -10383,7 +10389,7 @@ WITH batch AS (
|
||||
) AS last_model_config_id,
|
||||
(
|
||||
SELECT NULLIF(val, '')::chat_reasoning_effort
|
||||
FROM UNNEST($4::text[])
|
||||
FROM UNNEST($2::text[])
|
||||
WITH ORDINALITY AS t(val, ord)
|
||||
WHERE val != ''
|
||||
ORDER BY ord DESC
|
||||
@@ -10398,61 +10404,80 @@ updated_chat AS (
|
||||
last_reasoning_effort = COALESCE(batch.last_reasoning_effort, chats.last_reasoning_effort)
|
||||
FROM batch
|
||||
WHERE
|
||||
chats.id = $1::uuid
|
||||
chats.id = $3::uuid
|
||||
AND (
|
||||
chats.last_model_config_id IS DISTINCT FROM COALESCE(batch.last_model_config_id, chats.last_model_config_id)
|
||||
OR chats.last_reasoning_effort IS DISTINCT FROM COALESCE(batch.last_reasoning_effort, chats.last_reasoning_effort)
|
||||
)
|
||||
),
|
||||
allocated AS MATERIALIZED (
|
||||
-- Numbering the ids by value, rather than by the order nextval produced
|
||||
-- them, is what makes ordinal k always the k-th smallest id. MATERIALIZED
|
||||
-- is redundant while nextval is volatile, and pins that if it changes.
|
||||
SELECT
|
||||
id,
|
||||
(ROW_NUMBER() OVER (ORDER BY id))::int AS ord
|
||||
FROM (
|
||||
SELECT nextval('chat_messages_id_seq') AS id
|
||||
FROM generate_series(1, cardinality($4::chat_message_role[]))
|
||||
) s
|
||||
),
|
||||
inserted AS (
|
||||
INSERT INTO chat_messages (
|
||||
id,
|
||||
chat_id,
|
||||
created_by,
|
||||
model_config_id,
|
||||
reasoning_effort,
|
||||
role,
|
||||
content,
|
||||
content_version,
|
||||
visibility,
|
||||
input_tokens,
|
||||
output_tokens,
|
||||
total_tokens,
|
||||
reasoning_tokens,
|
||||
cache_creation_tokens,
|
||||
cache_read_tokens,
|
||||
context_limit,
|
||||
compressed,
|
||||
total_cost_micros,
|
||||
runtime_ms
|
||||
)
|
||||
SELECT
|
||||
allocated.id,
|
||||
$3::uuid,
|
||||
NULLIF(($5::uuid[])[allocated.ord], '00000000-0000-0000-0000-000000000000'::uuid),
|
||||
NULLIF(($1::uuid[])[allocated.ord], '00000000-0000-0000-0000-000000000000'::uuid),
|
||||
NULLIF(($2::text[])[allocated.ord], '')::chat_reasoning_effort,
|
||||
($4::chat_message_role[])[allocated.ord],
|
||||
($6::text[])[allocated.ord]::jsonb,
|
||||
($7::smallint[])[allocated.ord],
|
||||
($8::chat_message_visibility[])[allocated.ord],
|
||||
NULLIF(($9::bigint[])[allocated.ord], 0),
|
||||
NULLIF(($10::bigint[])[allocated.ord], 0),
|
||||
NULLIF(($11::bigint[])[allocated.ord], 0),
|
||||
NULLIF(($12::bigint[])[allocated.ord], 0),
|
||||
NULLIF(($13::bigint[])[allocated.ord], 0),
|
||||
NULLIF(($14::bigint[])[allocated.ord], 0),
|
||||
NULLIF(($15::bigint[])[allocated.ord], 0),
|
||||
($16::boolean[])[allocated.ord],
|
||||
NULLIF(($17::bigint[])[allocated.ord], 0),
|
||||
NULLIF(($18::bigint[])[allocated.ord], 0)
|
||||
FROM allocated
|
||||
RETURNING id, chat_id, model_config_id, created_at, role, content, visibility, input_tokens, output_tokens, total_tokens, reasoning_tokens, cache_creation_tokens, cache_read_tokens, context_limit, compressed, created_by, content_version, total_cost_micros, runtime_ms, deleted, provider_response_id, revision, reasoning_effort, search_tsv
|
||||
)
|
||||
INSERT INTO chat_messages (
|
||||
chat_id,
|
||||
created_by,
|
||||
model_config_id,
|
||||
reasoning_effort,
|
||||
role,
|
||||
content,
|
||||
content_version,
|
||||
visibility,
|
||||
input_tokens,
|
||||
output_tokens,
|
||||
total_tokens,
|
||||
reasoning_tokens,
|
||||
cache_creation_tokens,
|
||||
cache_read_tokens,
|
||||
context_limit,
|
||||
compressed,
|
||||
total_cost_micros,
|
||||
runtime_ms
|
||||
)
|
||||
SELECT
|
||||
$1::uuid,
|
||||
NULLIF(UNNEST($2::uuid[]), '00000000-0000-0000-0000-000000000000'::uuid),
|
||||
NULLIF(UNNEST($3::uuid[]), '00000000-0000-0000-0000-000000000000'::uuid),
|
||||
NULLIF(UNNEST($4::text[]), '')::chat_reasoning_effort,
|
||||
UNNEST($5::chat_message_role[]),
|
||||
UNNEST($6::text[])::jsonb,
|
||||
UNNEST($7::smallint[]),
|
||||
UNNEST($8::chat_message_visibility[]),
|
||||
NULLIF(UNNEST($9::bigint[]), 0),
|
||||
NULLIF(UNNEST($10::bigint[]), 0),
|
||||
NULLIF(UNNEST($11::bigint[]), 0),
|
||||
NULLIF(UNNEST($12::bigint[]), 0),
|
||||
NULLIF(UNNEST($13::bigint[]), 0),
|
||||
NULLIF(UNNEST($14::bigint[]), 0),
|
||||
NULLIF(UNNEST($15::bigint[]), 0),
|
||||
UNNEST($16::boolean[]),
|
||||
NULLIF(UNNEST($17::bigint[]), 0),
|
||||
NULLIF(UNNEST($18::bigint[]), 0)
|
||||
RETURNING
|
||||
id, chat_id, model_config_id, created_at, role, content, visibility, input_tokens, output_tokens, total_tokens, reasoning_tokens, cache_creation_tokens, cache_read_tokens, context_limit, compressed, created_by, content_version, total_cost_micros, runtime_ms, deleted, provider_response_id, revision, reasoning_effort, search_tsv
|
||||
SELECT id, chat_id, model_config_id, created_at, role, content, visibility, input_tokens, output_tokens, total_tokens, reasoning_tokens, cache_creation_tokens, cache_read_tokens, context_limit, compressed, created_by, content_version, total_cost_micros, runtime_ms, deleted, provider_response_id, revision, reasoning_effort, search_tsv
|
||||
FROM inserted
|
||||
ORDER BY id
|
||||
`
|
||||
|
||||
type InsertChatMessagesParams struct {
|
||||
ChatID uuid.UUID `db:"chat_id" json:"chat_id"`
|
||||
CreatedBy []uuid.UUID `db:"created_by" json:"created_by"`
|
||||
ModelConfigID []uuid.UUID `db:"model_config_id" json:"model_config_id"`
|
||||
ReasoningEffort []string `db:"reasoning_effort" json:"reasoning_effort"`
|
||||
ChatID uuid.UUID `db:"chat_id" json:"chat_id"`
|
||||
Role []ChatMessageRole `db:"role" json:"role"`
|
||||
CreatedBy []uuid.UUID `db:"created_by" json:"created_by"`
|
||||
Content []string `db:"content" json:"content"`
|
||||
ContentVersion []int16 `db:"content_version" json:"content_version"`
|
||||
Visibility []ChatMessageVisibility `db:"visibility" json:"visibility"`
|
||||
@@ -10468,13 +10493,43 @@ type InsertChatMessagesParams struct {
|
||||
RuntimeMs []int64 `db:"runtime_ms" json:"runtime_ms"`
|
||||
}
|
||||
|
||||
func (q *sqlQuerier) InsertChatMessages(ctx context.Context, arg InsertChatMessagesParams) ([]ChatMessage, error) {
|
||||
type InsertChatMessagesRow struct {
|
||||
ID int64 `db:"id" json:"id"`
|
||||
ChatID uuid.UUID `db:"chat_id" json:"chat_id"`
|
||||
ModelConfigID uuid.NullUUID `db:"model_config_id" json:"model_config_id"`
|
||||
CreatedAt time.Time `db:"created_at" json:"created_at"`
|
||||
Role ChatMessageRole `db:"role" json:"role"`
|
||||
Content pqtype.NullRawMessage `db:"content" json:"content"`
|
||||
Visibility ChatMessageVisibility `db:"visibility" json:"visibility"`
|
||||
InputTokens sql.NullInt64 `db:"input_tokens" json:"input_tokens"`
|
||||
OutputTokens sql.NullInt64 `db:"output_tokens" json:"output_tokens"`
|
||||
TotalTokens sql.NullInt64 `db:"total_tokens" json:"total_tokens"`
|
||||
ReasoningTokens sql.NullInt64 `db:"reasoning_tokens" json:"reasoning_tokens"`
|
||||
CacheCreationTokens sql.NullInt64 `db:"cache_creation_tokens" json:"cache_creation_tokens"`
|
||||
CacheReadTokens sql.NullInt64 `db:"cache_read_tokens" json:"cache_read_tokens"`
|
||||
ContextLimit sql.NullInt64 `db:"context_limit" json:"context_limit"`
|
||||
Compressed bool `db:"compressed" json:"compressed"`
|
||||
CreatedBy uuid.NullUUID `db:"created_by" json:"created_by"`
|
||||
ContentVersion int16 `db:"content_version" json:"content_version"`
|
||||
TotalCostMicros sql.NullInt64 `db:"total_cost_micros" json:"total_cost_micros"`
|
||||
RuntimeMs sql.NullInt64 `db:"runtime_ms" json:"runtime_ms"`
|
||||
Deleted bool `db:"deleted" json:"deleted"`
|
||||
ProviderResponseID sql.NullString `db:"provider_response_id" json:"provider_response_id"`
|
||||
Revision int64 `db:"revision" json:"revision"`
|
||||
ReasoningEffort NullChatReasoningEffort `db:"reasoning_effort" json:"reasoning_effort"`
|
||||
SearchTsv interface{} `db:"search_tsv" json:"search_tsv"`
|
||||
}
|
||||
|
||||
// Returns the inserted rows in input array order. Ids are allocated before the
|
||||
// insert and the k-th smallest is assigned to input index k, so callers may
|
||||
// index the result positionally.
|
||||
func (q *sqlQuerier) InsertChatMessages(ctx context.Context, arg InsertChatMessagesParams) ([]InsertChatMessagesRow, error) {
|
||||
rows, err := q.db.QueryContext(ctx, insertChatMessages,
|
||||
arg.ChatID,
|
||||
pq.Array(arg.CreatedBy),
|
||||
pq.Array(arg.ModelConfigID),
|
||||
pq.Array(arg.ReasoningEffort),
|
||||
arg.ChatID,
|
||||
pq.Array(arg.Role),
|
||||
pq.Array(arg.CreatedBy),
|
||||
pq.Array(arg.Content),
|
||||
pq.Array(arg.ContentVersion),
|
||||
pq.Array(arg.Visibility),
|
||||
@@ -10493,9 +10548,9 @@ func (q *sqlQuerier) InsertChatMessages(ctx context.Context, arg InsertChatMessa
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var items []ChatMessage
|
||||
var items []InsertChatMessagesRow
|
||||
for rows.Next() {
|
||||
var i ChatMessage
|
||||
var i InsertChatMessagesRow
|
||||
if err := rows.Scan(
|
||||
&i.ID,
|
||||
&i.ChatID,
|
||||
|
||||
Reference in New Issue
Block a user