opensource/coder
3
0
Fork 0
mirror of https://github.com/coder/coder.git synced 2026-09-24 15:04:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: replace pgcoordinator pg_notify triggers with app-level Publish() (#24717)

Browse Source
This commit is contained in:
Jon Ayers
2026-05-01 15:00:08 -05:00
committed by GitHub
parent 453a39be97
commit 6b9637d85a
12 changed files with 201 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
coderd/database/dbauthz/dbauthz.go
+4 -4
View File
@@ -1853,9 +1853,9 @@ func (q *querier) DeleteAllChatQueuedMessages(ctx context.Context, chatID uuid.U
return q.db.DeleteAllChatQueuedMessages(ctx, chatID)
}
func (q *querier) DeleteAllTailnetTunnels(ctx context.Context, arg database.DeleteAllTailnetTunnelsParams) error {
func (q *querier) DeleteAllTailnetTunnels(ctx context.Context, arg database.DeleteAllTailnetTunnelsParams) ([]database.DeleteAllTailnetTunnelsRow, error) {
if err := q.authorizeContext(ctx, policy.ActionDelete, rbac.ResourceTailnetCoordinator); err != nil {
return err
return nil, err
}
return q.db.DeleteAllTailnetTunnels(ctx, arg)
}
@@ -6676,9 +6676,9 @@ func (q *querier) UpdateReplica(ctx context.Context, arg database.UpdateReplicaP
return q.db.UpdateReplica(ctx, arg)
}
func (q *querier) UpdateTailnetPeerStatusByCoordinator(ctx context.Context, arg database.UpdateTailnetPeerStatusByCoordinatorParams) error {
func (q *querier) UpdateTailnetPeerStatusByCoordinator(ctx context.Context, arg database.UpdateTailnetPeerStatusByCoordinatorParams) ([]uuid.UUID, error) {
if err := q.authorizeContext(ctx, policy.ActionUpdate, rbac.ResourceTailnetCoordinator); err != nil {
return err
return nil, err
}
return q.db.UpdateTailnetPeerStatusByCoordinator(ctx, arg)
}
coderd/database/dbmetrics/querymetrics.go
+6 -6
View File
@@ -400,12 +400,12 @@ func (m queryMetricsStore) DeleteAllChatQueuedMessages(ctx context.Context, chat
return r0
}
func (m queryMetricsStore) DeleteAllTailnetTunnels(ctx context.Context, arg database.DeleteAllTailnetTunnelsParams) error {
func (m queryMetricsStore) DeleteAllTailnetTunnels(ctx context.Context, arg database.DeleteAllTailnetTunnelsParams) ([]database.DeleteAllTailnetTunnelsRow, error) {
start := time.Now()
r0 := m.s.DeleteAllTailnetTunnels(ctx, arg)
r0, r1 := m.s.DeleteAllTailnetTunnels(ctx, arg)
m.queryLatencies.WithLabelValues("DeleteAllTailnetTunnels").Observe(time.Since(start).Seconds())
m.queryCounts.WithLabelValues(httpmw.ExtractHTTPRoute(ctx), httpmw.ExtractHTTPMethod(ctx), "DeleteAllTailnetTunnels").Inc()
return r0
return r0, r1
}
func (m queryMetricsStore) DeleteAllWebpushSubscriptions(ctx context.Context) error {
@@ -4784,12 +4784,12 @@ func (m queryMetricsStore) UpdateReplica(ctx context.Context, arg database.Updat
return r0, r1
}
func (m queryMetricsStore) UpdateTailnetPeerStatusByCoordinator(ctx context.Context, arg database.UpdateTailnetPeerStatusByCoordinatorParams) error {
func (m queryMetricsStore) UpdateTailnetPeerStatusByCoordinator(ctx context.Context, arg database.UpdateTailnetPeerStatusByCoordinatorParams) ([]uuid.UUID, error) {
start := time.Now()
r0 := m.s.UpdateTailnetPeerStatusByCoordinator(ctx, arg)
r0, r1 := m.s.UpdateTailnetPeerStatusByCoordinator(ctx, arg)
m.queryLatencies.WithLabelValues("UpdateTailnetPeerStatusByCoordinator").Observe(time.Since(start).Seconds())
m.queryCounts.WithLabelValues(httpmw.ExtractHTTPRoute(ctx), httpmw.ExtractHTTPMethod(ctx), "UpdateTailnetPeerStatusByCoordinator").Inc()
return r0
return r0, r1
}
func (m queryMetricsStore) UpdateTaskPrompt(ctx context.Context, arg database.UpdateTaskPromptParams) (database.TaskTable, error) {
coderd/database/dbmock/dbmock.go
+8 -6
View File
@@ -645,11 +645,12 @@ func (mr *MockStoreMockRecorder) DeleteAllChatQueuedMessages(ctx, chatID any) *g
}
// DeleteAllTailnetTunnels mocks base method.
func (m *MockStore) DeleteAllTailnetTunnels(ctx context.Context, arg database.DeleteAllTailnetTunnelsParams) error {
func (m *MockStore) DeleteAllTailnetTunnels(ctx context.Context, arg database.DeleteAllTailnetTunnelsParams) ([]database.DeleteAllTailnetTunnelsRow, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "DeleteAllTailnetTunnels", ctx, arg)
ret0, _ := ret[0].(error)
return ret0
ret0, _ := ret[0].([]database.DeleteAllTailnetTunnelsRow)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// DeleteAllTailnetTunnels indicates an expected call of DeleteAllTailnetTunnels.
@@ -9033,11 +9034,12 @@ func (mr *MockStoreMockRecorder) UpdateReplica(ctx, arg any) *gomock.Call {
}
// UpdateTailnetPeerStatusByCoordinator mocks base method.
func (m *MockStore) UpdateTailnetPeerStatusByCoordinator(ctx context.Context, arg database.UpdateTailnetPeerStatusByCoordinatorParams) error {
func (m *MockStore) UpdateTailnetPeerStatusByCoordinator(ctx context.Context, arg database.UpdateTailnetPeerStatusByCoordinatorParams) ([]uuid.UUID, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "UpdateTailnetPeerStatusByCoordinator", ctx, arg)
ret0, _ := ret[0].(error)
return ret0
ret0, _ := ret[0].([]uuid.UUID)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// UpdateTailnetPeerStatusByCoordinator indicates an expected call of UpdateTailnetPeerStatusByCoordinator.
coderd/database/dump.sql
Generated
-44
View File
@@ -1061,44 +1061,6 @@ BEGIN
END;
$$;
CREATE FUNCTION tailnet_notify_coordinator_heartbeat() RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
PERFORM pg_notify('tailnet_coordinator_heartbeat', NEW.id::text);
RETURN NULL;
END;
$$;
CREATE FUNCTION tailnet_notify_peer_change() RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
IF (OLD IS NOT NULL) THEN
PERFORM pg_notify('tailnet_peer_update', OLD.id::text);
RETURN NULL;
END IF;
IF (NEW IS NOT NULL) THEN
PERFORM pg_notify('tailnet_peer_update', NEW.id::text);
RETURN NULL;
END IF;
END;
$$;
CREATE FUNCTION tailnet_notify_tunnel_change() RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
IF (NEW IS NOT NULL) THEN
PERFORM pg_notify('tailnet_tunnel_update', NEW.src_id || ',' || NEW.dst_id);
RETURN NULL;
ELSIF (OLD IS NOT NULL) THEN
PERFORM pg_notify('tailnet_tunnel_update', OLD.src_id || ',' || OLD.dst_id);
RETURN NULL;
END IF;
END;
$$;
CREATE TABLE ai_seat_state (
user_id uuid NOT NULL,
first_used_at timestamp with time zone NOT NULL,
@@ -4101,12 +4063,6 @@ CREATE TRIGGER remove_organization_member_custom_role BEFORE DELETE ON custom_ro
COMMENT ON TRIGGER remove_organization_member_custom_role ON custom_roles IS 'When a custom_role is deleted, this trigger removes the role from all organization members.';
CREATE TRIGGER tailnet_notify_coordinator_heartbeat AFTER INSERT OR UPDATE ON tailnet_coordinators FOR EACH ROW EXECUTE FUNCTION tailnet_notify_coordinator_heartbeat();
CREATE TRIGGER tailnet_notify_peer_change AFTER INSERT OR DELETE OR UPDATE ON tailnet_peers FOR EACH ROW EXECUTE FUNCTION tailnet_notify_peer_change();
CREATE TRIGGER tailnet_notify_tunnel_change AFTER INSERT OR DELETE OR UPDATE ON tailnet_tunnels FOR EACH ROW EXECUTE FUNCTION tailnet_notify_tunnel_change();
CREATE TRIGGER trigger_aggregate_usage_event AFTER INSERT ON usage_events FOR EACH ROW EXECUTE FUNCTION aggregate_usage_event();
CREATE TRIGGER trigger_delete_group_members_on_org_member_delete BEFORE DELETE ON organization_members FOR EACH ROW EXECUTE FUNCTION delete_group_members_on_org_member_delete();
coderd/database/migrations/000483_drop_tailnet_notify_triggers.down.sql
+43
View File
@@ -0,0 +1,43 @@
CREATE FUNCTION tailnet_notify_coordinator_heartbeat() RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
PERFORM pg_notify('tailnet_coordinator_heartbeat', NEW.id::text);
RETURN NULL;
END;
$$;
CREATE FUNCTION tailnet_notify_peer_change() RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
IF (OLD IS NOT NULL) THEN
PERFORM pg_notify('tailnet_peer_update', OLD.id::text);
RETURN NULL;
END IF;
IF (NEW IS NOT NULL) THEN
PERFORM pg_notify('tailnet_peer_update', NEW.id::text);
RETURN NULL;
END IF;
END;
$$;
CREATE FUNCTION tailnet_notify_tunnel_change() RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
IF (NEW IS NOT NULL) THEN
PERFORM pg_notify('tailnet_tunnel_update', NEW.src_id || ',' || NEW.dst_id);
RETURN NULL;
ELSIF (OLD IS NOT NULL) THEN
PERFORM pg_notify('tailnet_tunnel_update', OLD.src_id || ',' || OLD.dst_id);
RETURN NULL;
END IF;
END;
$$;
CREATE TRIGGER tailnet_notify_coordinator_heartbeat AFTER INSERT OR UPDATE ON tailnet_coordinators FOR EACH ROW EXECUTE FUNCTION tailnet_notify_coordinator_heartbeat();
CREATE TRIGGER tailnet_notify_peer_change AFTER INSERT OR DELETE OR UPDATE ON tailnet_peers FOR EACH ROW EXECUTE FUNCTION tailnet_notify_peer_change();
CREATE TRIGGER tailnet_notify_tunnel_change AFTER INSERT OR DELETE OR UPDATE ON tailnet_tunnels FOR EACH ROW EXECUTE FUNCTION tailnet_notify_tunnel_change();
coderd/database/migrations/000483_drop_tailnet_notify_triggers.up.sql
+6
View File
@@ -0,0 +1,6 @@
DROP TRIGGER IF EXISTS tailnet_notify_peer_change ON tailnet_peers;
DROP TRIGGER IF EXISTS tailnet_notify_tunnel_change ON tailnet_tunnels;
DROP TRIGGER IF EXISTS tailnet_notify_coordinator_heartbeat ON tailnet_coordinators;
DROP FUNCTION IF EXISTS tailnet_notify_peer_change();
DROP FUNCTION IF EXISTS tailnet_notify_tunnel_change();
DROP FUNCTION IF EXISTS tailnet_notify_coordinator_heartbeat();
coderd/database/querier.go
+2 -2
View File
@@ -101,7 +101,7 @@ type sqlcQuerier interface {
DeleteAPIKeyByID(ctx context.Context, id string) error
DeleteAPIKeysByUserID(ctx context.Context, userID uuid.UUID) error
DeleteAllChatQueuedMessages(ctx context.Context, chatID uuid.UUID) error
DeleteAllTailnetTunnels(ctx context.Context, arg DeleteAllTailnetTunnelsParams) error
DeleteAllTailnetTunnels(ctx context.Context, arg DeleteAllTailnetTunnelsParams) ([]DeleteAllTailnetTunnelsRow, error)
// Deletes all existing webpush subscriptions.
// This should be called when the VAPID keypair is regenerated, as the old
// keypair will no longer be valid and all existing subscriptions will need to
@@ -1117,7 +1117,7 @@ type sqlcQuerier interface {
UpdateProvisionerJobWithCompleteByID(ctx context.Context, arg UpdateProvisionerJobWithCompleteByIDParams) error
UpdateProvisionerJobWithCompleteWithStartedAtByID(ctx context.Context, arg UpdateProvisionerJobWithCompleteWithStartedAtByIDParams) error
UpdateReplica(ctx context.Context, arg UpdateReplicaParams) (Replica, error)
UpdateTailnetPeerStatusByCoordinator(ctx context.Context, arg UpdateTailnetPeerStatusByCoordinatorParams) error
UpdateTailnetPeerStatusByCoordinator(ctx context.Context, arg UpdateTailnetPeerStatusByCoordinatorParams) ([]uuid.UUID, error)
UpdateTaskPrompt(ctx context.Context, arg UpdateTaskPromptParams) (TaskTable, error)
UpdateTaskWorkspaceID(ctx context.Context, arg UpdateTaskWorkspaceIDParams) (TaskTable, error)
UpdateTemplateACLByID(ctx context.Context, arg UpdateTemplateACLByIDParams) error
coderd/database/queries.sql.go
+51 -8
View File
@@ -21263,10 +21263,11 @@ func (q *sqlQuerier) CleanTailnetTunnels(ctx context.Context) error {
return err
}
const deleteAllTailnetTunnels = `-- name: DeleteAllTailnetTunnels :exec
const deleteAllTailnetTunnels = `-- name: DeleteAllTailnetTunnels :many
DELETE
FROM tailnet_tunnels
WHERE coordinator_id = $1 and src_id = $2
RETURNING src_id, dst_id
`
type DeleteAllTailnetTunnelsParams struct {
@@ -21274,9 +21275,32 @@ type DeleteAllTailnetTunnelsParams struct {
SrcID uuid.UUID `db:"src_id" json:"src_id"`
}
func (q *sqlQuerier) DeleteAllTailnetTunnels(ctx context.Context, arg DeleteAllTailnetTunnelsParams) error {
_, err := q.db.ExecContext(ctx, deleteAllTailnetTunnels, arg.CoordinatorID, arg.SrcID)
return err
type DeleteAllTailnetTunnelsRow struct {
SrcID uuid.UUID `db:"src_id" json:"src_id"`
DstID uuid.UUID `db:"dst_id" json:"dst_id"`
}
func (q *sqlQuerier) DeleteAllTailnetTunnels(ctx context.Context, arg DeleteAllTailnetTunnelsParams) ([]DeleteAllTailnetTunnelsRow, error) {
rows, err := q.db.QueryContext(ctx, deleteAllTailnetTunnels, arg.CoordinatorID, arg.SrcID)
if err != nil {
return nil, err
}
defer rows.Close()
var items []DeleteAllTailnetTunnelsRow
for rows.Next() {
var i DeleteAllTailnetTunnelsRow
if err := rows.Scan(&i.SrcID, &i.DstID); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Close(); err != nil {
return nil, err
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const deleteTailnetPeer = `-- name: DeleteTailnetPeer :one
@@ -21551,13 +21575,14 @@ func (q *sqlQuerier) GetTailnetTunnelPeerIDsBatch(ctx context.Context, ids []uui
return items, nil
}
const updateTailnetPeerStatusByCoordinator = `-- name: UpdateTailnetPeerStatusByCoordinator :exec
const updateTailnetPeerStatusByCoordinator = `-- name: UpdateTailnetPeerStatusByCoordinator :many
UPDATE
tailnet_peers
SET
status = $2
WHERE
coordinator_id = $1
RETURNING id
`
type UpdateTailnetPeerStatusByCoordinatorParams struct {
@@ -21565,9 +21590,27 @@ type UpdateTailnetPeerStatusByCoordinatorParams struct {
Status TailnetStatus `db:"status" json:"status"`
}
func (q *sqlQuerier) UpdateTailnetPeerStatusByCoordinator(ctx context.Context, arg UpdateTailnetPeerStatusByCoordinatorParams) error {
_, err := q.db.ExecContext(ctx, updateTailnetPeerStatusByCoordinator, arg.CoordinatorID, arg.Status)
return err
func (q *sqlQuerier) UpdateTailnetPeerStatusByCoordinator(ctx context.Context, arg UpdateTailnetPeerStatusByCoordinatorParams) ([]uuid.UUID, error) {
rows, err := q.db.QueryContext(ctx, updateTailnetPeerStatusByCoordinator, arg.CoordinatorID, arg.Status)
if err != nil {
return nil, err
}
defer rows.Close()
var items []uuid.UUID
for rows.Next() {
var id uuid.UUID
if err := rows.Scan(&id); err != nil {
return nil, err
}
items = append(items, id)
}
if err := rows.Close(); err != nil {
return nil, err
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const upsertTailnetCoordinator = `-- name: UpsertTailnetCoordinator :one
coderd/database/queries/tailnet.sql
+6 -4
View File
@@ -50,13 +50,14 @@ DO UPDATE SET
updated_at = now() at time zone 'utc'
RETURNING *;
-- name: UpdateTailnetPeerStatusByCoordinator :exec
-- name: UpdateTailnetPeerStatusByCoordinator :many
UPDATE
tailnet_peers
SET
status = $2
WHERE
coordinator_id = $1;
coordinator_id = $1
RETURNING id;
-- name: DeleteTailnetPeer :one
DELETE
@@ -91,10 +92,11 @@ FROM tailnet_tunnels
WHERE coordinator_id = $1 and src_id = $2 and dst_id = $3
RETURNING coordinator_id, src_id, dst_id;
-- name: DeleteAllTailnetTunnels :exec
-- name: DeleteAllTailnetTunnels :many
DELETE
FROM tailnet_tunnels
WHERE coordinator_id = $1 and src_id = $2;
WHERE coordinator_id = $1 and src_id = $2
RETURNING src_id, dst_id;
-- For PG Coordinator HTMLDebug