Skip to content

feat: clear migrated messages #30

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jenbutongit merged 5 commits into from
Oct 1, 2024
Merged

feat: clear migrated messages #30

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
b81f2bc
update circleCI images
jenbutongit Sep 30, 2024
52cc3c8
remove net-tools install
jenbutongit Sep 30, 2024
a280269
feat: delete old messages during migration
jenbutongit Oct 1, 2024
cad7af5
feat: migrate any in retry state. remove intermediary update
jenbutongit Oct 1, 2024
c3c8d8d
feat: add 'retry' to getInflightMessages query
jenbutongit Oct 1, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 2 additions & 3 deletions .circleci/config.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ jobs:
type: string

machine:
image: ubuntu-2004:202201-02
image: ubuntu-2004:202111-02

steps:
- attach_workspace:
Expand Down Expand Up @@ -114,8 +114,7 @@ jobs:
name: Install OpenVPN
command: |
sudo apt-get update
sudo apt-get install net-tools -y
sudo apt-get install openvpn openvpn-systemd-resolved -y
sudo apt-get install openvpn openvpn-systemd-resolved

- run:
name: Check IP before VPN connection
Expand Down
20 changes: 10 additions & 10 deletions worker/src/Consumer/migrate/helpers.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ export async function getInflightMessages(queue: string, schema: string) {
${schema}.job
WHERE
name = '${queue}'
AND state = 'created'`;
AND state in ('created', 'retry')`;

const inflightMessages = await db.query(query);
const numberOfMessages = inflightMessages?.rows.length;
Expand All @@ -45,8 +45,8 @@ export async function migrateInflightMessagesFromV9(queue: string, schema: strin
const client = await db.connect();

try {
logger.info(`Copying inflight messages for queue ${queue} from ${schema} to ${currentSchema} and deleting old messages`);
await client.query("BEGIN");

await client.query(`INSERT INTO ${currentSchema}.job (
id,
name,
Expand Down Expand Up @@ -74,16 +74,16 @@ export async function migrateInflightMessagesFromV9(queue: string, schema: strin
output
FROM ${schema}.job
WHERE name = '${queue}'
AND state = 'created'
AND state IN ('created', 'retry')
ON CONFLICT DO NOTHING`);

logger.info(`Copied inflight messages for queue ${queue} from ${schema} to ${currentSchema}`);

await client.query(`
UPDATE ${schema}.job
SET state = 'completed'
WHERE name = '${queue}' and state = 'created'
`);
DELETE from ${schema}.job
WHERE name = '${queue}' and state IN ('created', 'completed', 'retry');

DELETE from ${schema}.archive
WHERE name = '${queue}' and state IN ('created', 'completed', 'retry');
`);

await client.query("COMMIT");
} catch (err) {
Expand All @@ -94,5 +94,5 @@ export async function migrateInflightMessagesFromV9(queue: string, schema: strin
await client.release();
}

logger.info({ drainSchema: schema, queue }, `Inflight messages drained successfully for schema ${schema}, queue ${queue}.`);
logger.info({ drainSchema: schema, queue }, `Inflight messages drained successfully for queue ${queue}, from schema ${schema} to ${currentSchema}`);
}