calvin.erfmann/pangolin
1
0
Fork 0
mirror of https://github.com/fosrl/pangolin.git synced 2026-08-25 21:45:24 +02:00
Code Issues Packages Projects Releases Wiki Activity

Move the messaging out of the transaction

Browse Source
This commit is contained in:
Owen
2026-08-24 17:01:37 -04:00
parent d2809fbfd1
commit 23764feb4f
1 changed files with 85 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/lib/blueprints/applyBlueprint.ts
+85 -74
View File
@@ -1,5 +1,6 @@
import { import {
db, db,
primaryDb,
newts, newts,
blueprints, blueprints,
Blueprint, Blueprint,
@@ -80,93 +81,103 @@ export async function applyBlueprint({
trx, trx,
siteId siteId
); );
});
// We need to update the targets on the newts from the successfully updated information // Push updates to newts/clients only after the transaction has
for (const result of publicResourcesResults) { // committed. Doing this while the transaction is still open can
for (const target of result.targetsToUpdate) { // race with the writes (e.g. newts requesting config before the
const [site] = await trx // new targets/resources are actually visible), leaving them out
.select() // of sync until manually toggled.
.from(sites)
.innerJoin(newts, eq(sites.siteId, newts.siteId)) // We need to update the targets on the newts from the successfully updated information
.where( for (const result of publicResourcesResults) {
and( for (const target of result.targetsToUpdate) {
eq(sites.siteId, target.siteId), // read from the primary: this determines whether/how we push
eq(sites.orgId, orgId), // the just-created target to the newt, so a lagging replica
eq(sites.type, "newt"), // returning stale or missing data here would silently skip
isNotNull(sites.pubKey) // the push
) const [site] = await primaryDb
.select()
.from(sites)
.innerJoin(newts, eq(sites.siteId, newts.siteId))
.where(
and(
eq(sites.siteId, target.siteId),
eq(sites.orgId, orgId),
eq(sites.type, "newt"),
isNotNull(sites.pubKey)
) )
.limit(1); )
.limit(1);
if (site) { if (site) {
logger.debug( logger.debug(
`Updating target ${target.targetId} on site ${site.sites.siteId}` `Updating target ${target.targetId} on site ${site.sites.siteId}`
);
// see if you can find a matching target health check from the healthchecksToUpdate array
const matchingHealthcheck =
result.healthchecksToUpdate.find(
(hc) => hc.targetId === target.targetId
); );
// see if you can find a matching target health check from the healthchecksToUpdate array if (["http", "tcp", "udp"].includes(target.mode)) {
const matchingHealthcheck = await addProxyTargets(
result.healthchecksToUpdate.find( site.newt.newtId,
(hc) => hc.targetId === target.targetId [target],
); matchingHealthcheck
? [matchingHealthcheck]
if (["http", "tcp", "udp"].includes(target.mode)) { : [],
await addProxyTargets( result.proxyResource.mode === "udp"
site.newt.newtId, ? "udp"
[target], : "tcp",
matchingHealthcheck site.newt.version
? [matchingHealthcheck] );
: [], } else if (
result.proxyResource.mode === "udp" ["ssh", "rdp", "vnc"].includes(target.mode)
? "udp" ) {
: "tcp", await sendBrowserGatewayTargets(
site.newt.version site.newt.newtId,
); [target],
} else if ( site.newt.version
["ssh", "rdp", "vnc"].includes(target.mode) );
) {
await sendBrowserGatewayTargets(
site.newt.newtId,
[target],
site.newt.version
);
}
} }
} }
} }
}
logger.debug( logger.debug(
`Successfully updated public resources for org ${orgId}: ${JSON.stringify(publicResourcesResults)}` `Successfully updated public resources for org ${orgId}: ${JSON.stringify(publicResourcesResults)}`
); );
// We need to update the targets on the newts from the successfully updated information // We need to update the targets on the newts from the successfully updated information
for (const result of privateResourcesResults) { for (const result of privateResourcesResults) {
rebuildClientAssociationsFromSiteResource( rebuildClientAssociationsFromSiteResource(
result.newSiteResource result.newSiteResource
)
.then(() =>
waitForSiteResourceRebuildIdle(
result.newSiteResource.siteResourceId
)
) )
.then(() => .then(() =>
waitForSiteResourceRebuildIdle( handleMessagingForUpdatedSiteResource(
result.newSiteResource.siteResourceId result.oldSiteResource,
) result.newSiteResource,
result.oldSites.map((s) => s.siteId),
result.newSites.map((s) => s.siteId)
) )
.then(() => )
handleMessagingForUpdatedSiteResource( .catch((e) => {
result.oldSiteResource, logger.error(
result.newSiteResource, `Failed to rebuild and handle messaging for site resource ${result.newSiteResource.siteResourceId}. Error: ${e}`
result.oldSites.map((s) => s.siteId), );
result.newSites.map((s) => s.siteId) });
) }
)
.catch((e) => {
logger.error(
`Failed to rebuild and handle messaging for site resource ${result.newSiteResource.siteResourceId}. Error: ${e}`
);
});
}
logger.debug( logger.debug(
`Successfully updated private resources for org ${orgId}: ${JSON.stringify(privateResourcesResults)}` `Successfully updated private resources for org ${orgId}: ${JSON.stringify(privateResourcesResults)}`
); );
});
blueprintSucceeded = true; blueprintSucceeded = true;
blueprintMessage = "Blueprint applied successfully"; blueprintMessage = "Blueprint applied successfully";