Skip to content

Commit 3c26593

Browse files
pascal-botpresspascal-botpress
committed
feat: add a ReplicationSlot custom resource
1 parent 45ded21 commit 3c26593

8 files changed

Lines changed: 446 additions & 2 deletions

File tree

‎cdk-postgresql/lib/handler.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import { VError } from "verror";
44
import { handler as dbHandler } from "./database.handler";
55
import { handler as roleHandler } from "./role.handler";
66
import { handler as roleMembershipHandler } from "./role-membership.handler";
7+
import { handler as replicationSlotHandler } from "./replication-slot.handler";
78

89
export const handler = async (event: CloudFormationCustomResourceEvent) => {
910
switch (event.ResourceType) {
@@ -13,6 +14,8 @@ export const handler = async (event: CloudFormationCustomResourceEvent) => {
1314
return dbHandler(event);
1415
case "Custom::Postgresql-RoleMembership":
1516
return roleMembershipHandler(event);
17+
case "Custom::Postgresql-ReplicationSlot":
18+
return replicationSlotHandler(event);
1619
default:
1720
throw new VError(`unexpected ResourceType: ${event.ResourceType}`);
1821
}

‎cdk-postgresql/lib/index.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
export * from "./database";
22
export * from "./role";
33
export * from "./role-membership";
4+
export * from "./replication-slot";
45
export * from "./provider";

‎cdk-postgresql/lib/postgres.ts‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -105,3 +105,28 @@ export const revokeRoleMembership = async (props: {
105105
console.warn(thrown.message);
106106
}
107107
};
108+
109+
export const createReplicationSlot = async (props: {
110+
client: Client;
111+
name: string;
112+
plugin: string;
113+
}) => {
114+
const { client, name, plugin } = props;
115+
116+
await client.query("SELECT pg_create_logical_replication_slot($1, $2)", [
117+
name,
118+
plugin,
119+
]);
120+
};
121+
122+
export const dropReplicationSlot = async (props: {
123+
client: Client;
124+
name: string;
125+
}) => {
126+
const { client, name } = props;
127+
128+
await client.query(
129+
"SELECT pg_drop_replication_slot(slot_name) FROM pg_replication_slots WHERE slot_name = $1",
130+
[name]
131+
);
132+
};

‎cdk-postgresql/lib/replication-slot.handler.ts‎

Lines changed: 145 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,145 @@
1+
import {
2+
CloudFormationCustomResourceEvent,
3+
CloudFormationCustomResourceCreateEvent,
4+
CloudFormationCustomResourceUpdateEvent,
5+
CloudFormationCustomResourceDeleteEvent,
6+
} from "aws-lambda/trigger/cloudformation-custom-resource";
7+
8+
import { validateConnection, hashCode, getConnectedClient } from "./util";
9+
import { Connection } from "./lambda.types";
10+
import * as postgres from "./postgres";
11+
12+
interface Props {
13+
ServiceToken: string;
14+
Connection: Connection;
15+
Name: string;
16+
Plugin: string;
17+
RequiredPublication?: string;
18+
}
19+
20+
export const handler = async (event: CloudFormationCustomResourceEvent) => {
21+
switch (event.RequestType) {
22+
case "Create":
23+
return handleCreate(event);
24+
case "Update":
25+
return handleUpdate(event);
26+
case "Delete":
27+
return handleDelete(event);
28+
}
29+
};
30+
31+
const handleCreate = async (event: CloudFormationCustomResourceCreateEvent) => {
32+
const props = event.ResourceProperties as Props;
33+
validateProps(props);
34+
await createSlotIfAbsent(props);
35+
return {
36+
PhysicalResourceId: generatePhysicalId(props),
37+
};
38+
};
39+
40+
/**
41+
* A slot cannot be altered in place: its name is its identity and its plugin is
42+
* fixed at creation. A name change reports a new physical id, so CloudFormation
43+
* creates the new slot and then deletes the old resource. A plugin change under
44+
* the same name is refused, because the existing slot would silently keep its
45+
* old plugin.
46+
*/
47+
const handleUpdate = async (event: CloudFormationCustomResourceUpdateEvent) => {
48+
const props = event.ResourceProperties as Props;
49+
validateProps(props);
50+
51+
const oldProps = event.OldResourceProperties as Props;
52+
const physicalResourceId = generatePhysicalId(props);
53+
const isDifferentSlot = physicalResourceId != event.PhysicalResourceId;
54+
55+
if (isDifferentSlot) {
56+
await createSlotIfAbsent(props);
57+
return { PhysicalResourceId: physicalResourceId };
58+
}
59+
60+
if (props.Plugin != oldProps.Plugin) {
61+
throw new Error(
62+
`The plugin of replication slot "${props.Name}" cannot change in place; drop and recreate the slot deliberately instead`
63+
);
64+
}
65+
66+
return { PhysicalResourceId: physicalResourceId };
67+
};
68+
69+
const handleDelete = async (event: CloudFormationCustomResourceDeleteEvent) => {
70+
const props = event.ResourceProperties as Props;
71+
validateProps(props);
72+
73+
console.log("Dropping replication slot", props.Name);
74+
const client = await getConnectedClient(props.Connection);
75+
76+
try {
77+
await postgres.dropReplicationSlot({ client, name: props.Name });
78+
} finally {
79+
await client.end();
80+
}
81+
82+
return {};
83+
};
84+
85+
const validateProps = (props: Props) => {
86+
if (!("Connection" in props)) {
87+
throw "Connection property is required";
88+
}
89+
validateConnection(props.Connection);
90+
91+
if (!("Name" in props)) {
92+
throw "Name property is required";
93+
}
94+
if (!("Plugin" in props)) {
95+
throw "Plugin property is required";
96+
}
97+
};
98+
99+
const generatePhysicalId = (props: Props): string => {
100+
const { Host, Port } = props.Connection;
101+
const suffix = Math.abs(hashCode(`${Host}-${Port}-${props.Name}`));
102+
return `replication-slot-${suffix}`;
103+
};
104+
105+
const createSlotIfAbsent = async (props: Props) => {
106+
const client = await getConnectedClient(props.Connection);
107+
108+
try {
109+
if (props.RequiredPublication) {
110+
const { rows } = await client.query(
111+
"SELECT 1 FROM pg_publication WHERE pubname = $1",
112+
[props.RequiredPublication]
113+
);
114+
if (rows.length === 0) {
115+
// A slot created before its publication decodes from a catalog snapshot
116+
// in which the publication does not exist, and replication then fails
117+
// continuously, so refusing here turns a subtle runtime failure into a
118+
// loud deploy failure:
119+
throw new Error(
120+
`publication "${props.RequiredPublication}" does not exist in database "${props.Connection.Database}"; deploy the migration that creates it before this slot`
121+
);
122+
}
123+
}
124+
125+
// `pg_replication_slots` lists the whole cluster, so a slot of the same
126+
// name decoding another database must not be adopted as this one; letting
127+
// Postgres raise "replication slot already exists" is the loud failure:
128+
const { rows: existing } = await client.query(
129+
"SELECT 1 FROM pg_replication_slots WHERE slot_name = $1 AND database = current_database()",
130+
[props.Name]
131+
);
132+
if (existing.length === 0) {
133+
console.log("Creating replication slot", props.Name);
134+
await postgres.createReplicationSlot({
135+
client,
136+
name: props.Name,
137+
plugin: props.Plugin,
138+
});
139+
} else {
140+
console.log("Replication slot already exists", props.Name);
141+
}
142+
} finally {
143+
await client.end();
144+
}
145+
};

‎cdk-postgresql/lib/replication-slot.ts‎

Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,71 @@
1+
import { Construct } from "constructs";
2+
import * as cdk from "aws-cdk-lib";
3+
import { RemovalPolicy } from "aws-cdk-lib";
4+
import { Provider } from "./provider";
5+
6+
export interface ReplicationSlotProps {
7+
/**
8+
* Provider required to connect to the Postgresql server. Logical slots decode
9+
* one database, the one the provider's `database` prop names.
10+
*/
11+
provider: Provider;
12+
13+
/**
14+
* The name of the slot. Must be unique on the PostgreSQL server instance
15+
* where it is configured.
16+
*/
17+
name: string;
18+
19+
/**
20+
* The logical decoding plugin the slot is created with. It cannot change
21+
* after creation.
22+
*
23+
* @default - "pgoutput"
24+
*/
25+
plugin?: string;
26+
27+
/**
28+
* A publication that has to exist before the slot is created. A slot created
29+
* earlier decodes from a catalog snapshot in which the publication does not
30+
* exist, and replication fails continuously, so creation is refused until
31+
* the publication is there.
32+
*/
33+
requiredPublication?: string;
34+
35+
/**
36+
* Policy to apply when the slot is removed from this stack. A dropped slot
37+
* loses its position permanently, so stacks that replicate production data
38+
* should retain it.
39+
*
40+
* @default - The slot will be dropped.
41+
*/
42+
removalPolicy?: RemovalPolicy;
43+
}
44+
45+
/**
46+
* A PostgreSQL logical replication slot. The database retains write-ahead-log
47+
* data until the slot's consumer confirms having read it, which is what makes
48+
* replication resumable after the consumer restarts.
49+
*/
50+
export class ReplicationSlot extends Construct {
51+
constructor(scope: Construct, id: string, props: ReplicationSlotProps) {
52+
super(scope, id);
53+
54+
const { provider, name, plugin, requiredPublication, removalPolicy } = props;
55+
56+
const cr = new cdk.CustomResource(this, "CustomResource", {
57+
serviceToken: provider.serviceToken,
58+
resourceType: "Custom::Postgresql-ReplicationSlot",
59+
properties: {
60+
connection: provider.buildConnectionProperty(),
61+
name,
62+
plugin: plugin ?? "pgoutput",
63+
requiredPublication,
64+
},
65+
pascalCaseProperties: true,
66+
});
67+
68+
cr.applyRemovalPolicy(removalPolicy || cdk.RemovalPolicy.DESTROY);
69+
cr.node.addDependency(provider);
70+
}
71+
}

‎cdk-postgresql/package.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
{
22
"name": "@botpress/cdk-postgresql",
3-
"version": "3.0.0",
3+
"version": "3.1.0",
44
"description": "Postgresql constructs for AWS CDK",
55
"main": "./dist/index.cjs",
66
"types": "./dist/index.d.cts",

‎cdk-postgresql/test/helpers.ts‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,3 +81,14 @@ export const isMemberOf = async (props: {
8181
);
8282
return rows[0].is_member;
8383
};
84+
85+
export const replicationSlotExists = async (
86+
client: Client,
87+
name: string
88+
): Promise<boolean> => {
89+
const { rows } = await client.query(
90+
"SELECT 1 FROM pg_replication_slots WHERE slot_name = $1",
91+
[name]
92+
);
93+
return rows.length > 0;
94+
};

0 commit comments

Comments
 (0)