calvin.erfmann/pangolin
1
0
Fork 0
mirror of https://github.com/fosrl/pangolin.git synced 2026-08-11 06:58:28 +02:00
Code Issues Packages Projects Releases Wiki Activity

Add some retry and database confict mitigation

Browse Source
This commit is contained in:
Owen
2026-07-03 10:23:32 -04:00
parent 1b1fba60f1
commit b399d2a291
3 changed files with 145 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/lib/rebuildClientAssociations.ts
+142 -115
View File
@@ -485,6 +485,7 @@ async function rebuildClientAssociationsFromSiteResourceImpl(
await trx await trx
.insert(clientSiteResourcesAssociationsCache) .insert(clientSiteResourcesAssociationsCache)
.values(clientSiteResourcesToInsert) .values(clientSiteResourcesToInsert)
.onConflictDoNothing()
.returning(); .returning();
logger.debug( logger.debug(
`rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteResourceId=${siteResource.siteResourceId} inserted clientSiteResource associations` `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteResourceId=${siteResource.siteResourceId} inserted clientSiteResource associations`
@@ -532,121 +533,141 @@ async function rebuildClientAssociationsFromSiteResourceImpl(
for (const site of sitesToProcess) { for (const site of sitesToProcess) {
const siteId = site.siteId; const siteId = site.siteId;
logger.debug( try {
`rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] processing siteId=${siteId} for siteResourceId=${siteResource.siteResourceId}` logger.debug(
); `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] processing siteId=${siteId} for siteResourceId=${siteResource.siteResourceId}`
);
const existingClientSites = await trx const existingClientSites = await trx
.select({ .select({
clientId: clientSitesAssociationsCache.clientId clientId: clientSitesAssociationsCache.clientId
}) })
.from(clientSitesAssociationsCache) .from(clientSitesAssociationsCache)
.where(eq(clientSitesAssociationsCache.siteId, siteId)); .where(eq(clientSitesAssociationsCache.siteId, siteId));
const existingClientSiteIds = existingClientSites.map( const existingClientSiteIds = existingClientSites.map(
(row) => row.clientId (row) => row.clientId
); );
logger.debug( logger.debug(
`rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} existingClientSiteIds=[${existingClientSiteIds.join(", ")}]` `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} existingClientSiteIds=[${existingClientSiteIds.join(", ")}]`
); );
// Get full client details for existing clients (needed for sending delete messages) // Get full client details for existing clients (needed for sending delete messages)
const existingClients = const existingClients =
existingClientSiteIds.length > 0 existingClientSiteIds.length > 0
? await trx ? await trx
.select({ .select({
clientId: clients.clientId, clientId: clients.clientId,
pubKey: clients.pubKey, pubKey: clients.pubKey,
subnet: clients.subnet subnet: clients.subnet
}) })
.from(clients) .from(clients)
.where(inArray(clients.clientId, existingClientSiteIds)) .where(
inArray(clients.clientId, existingClientSiteIds)
)
: [];
const otherResourceClientIds =
clientsFromOtherResourcesBySite.get(siteId) ??
new Set<number>();
logger.debug(
`rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} otherResourceClientIds=[${[...otherResourceClientIds].join(", ")}] mergedAllClientIds=[${mergedAllClientIds.join(", ")}]`
);
// Expected clients from this resource are site-scoped: if this site is
// no longer attached to the resource, the expected set is empty.
const expectedClientIdsForSite = currentSiteIdSet.has(siteId)
? mergedAllClientIds
: []; : [];
const otherResourceClientIds = const clientSitesToAdd = expectedClientIdsForSite.filter(
clientsFromOtherResourcesBySite.get(siteId) ?? new Set<number>(); (clientId) =>
!existingClientSiteIds.includes(clientId) &&
logger.debug( !otherResourceClientIds.has(clientId) // dont add if already connected via another site resource
`rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} otherResourceClientIds=[${[...otherResourceClientIds].join(", ")}] mergedAllClientIds=[${mergedAllClientIds.join(", ")}]`
);
// Expected clients from this resource are site-scoped: if this site is
// no longer attached to the resource, the expected set is empty.
const expectedClientIdsForSite = currentSiteIdSet.has(siteId)
? mergedAllClientIds
: [];
const clientSitesToAdd = expectedClientIdsForSite.filter(
(clientId) =>
!existingClientSiteIds.includes(clientId) &&
!otherResourceClientIds.has(clientId) // dont add if already connected via another site resource
);
const clientSitesToInsert = clientSitesToAdd.map((clientId) => ({
clientId,
siteId
}));
logger.debug(
`rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} clientSites toAdd=[${clientSitesToAdd.join(", ")}]`
);
if (clientSitesToInsert.length > 0) {
logger.debug(
`rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} inserting ${clientSitesToInsert.length} clientSite association(s)`
); );
await trx
.insert(clientSitesAssociationsCache)
.values(clientSitesToInsert)
.returning();
logger.debug(
`rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} inserted clientSite associations`
);
} else {
logger.debug(
`rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} no clientSite associations to insert`
);
}
// Now remove any client-site associations that should no longer exist const clientSitesToInsert = clientSitesToAdd.map((clientId) => ({
const clientSitesToRemove = existingClientSiteIds.filter( clientId,
(clientId) => siteId
!expectedClientIdsForSite.includes(clientId) && }));
!otherResourceClientIds.has(clientId) // dont remove if there is still another connection for another site resource
);
logger.debug(
`rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} clientSites toRemove=[${clientSitesToRemove.join(", ")}]`
);
if (clientSitesToRemove.length > 0) {
logger.debug( logger.debug(
`rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} deleting ${clientSitesToRemove.length} clientSite association(s)` `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} clientSites toAdd=[${clientSitesToAdd.join(", ")}]`
); );
await trx
.delete(clientSitesAssociationsCache) if (clientSitesToInsert.length > 0) {
.where( logger.debug(
and( `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} inserting ${clientSitesToInsert.length} clientSite association(s)`
eq(clientSitesAssociationsCache.siteId, siteId),
inArray(
clientSitesAssociationsCache.clientId,
clientSitesToRemove
)
)
); );
} await trx
.insert(clientSitesAssociationsCache)
.values(clientSitesToInsert)
.onConflictDoNothing()
.returning();
logger.debug(
`rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} inserted clientSite associations`
);
} else {
logger.debug(
`rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} no clientSite associations to insert`
);
}
// Now handle the messages to add/remove peers on both the newt and olm sides // Now remove any client-site associations that should no longer exist
await handleMessagesForSiteClients( const clientSitesToRemove = existingClientSiteIds.filter(
site, (clientId) =>
siteId, !expectedClientIdsForSite.includes(clientId) &&
mergedAllClients, !otherResourceClientIds.has(clientId) // dont remove if there is still another connection for another site resource
existingClients, );
clientSitesToAdd,
clientSitesToRemove, logger.debug(
trx `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} clientSites toRemove=[${clientSitesToRemove.join(", ")}]`
); );
if (clientSitesToRemove.length > 0) {
logger.debug(
`rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} deleting ${clientSitesToRemove.length} clientSite association(s)`
);
await trx
.delete(clientSitesAssociationsCache)
.where(
and(
eq(clientSitesAssociationsCache.siteId, siteId),
inArray(
clientSitesAssociationsCache.clientId,
clientSitesToRemove
)
)
);
}
// Now handle the messages to add/remove peers on both the newt and olm sides
await handleMessagesForSiteClients(
site,
siteId,
mergedAllClients,
existingClients,
clientSitesToAdd,
clientSitesToRemove,
trx
);
} catch (err) {
// Don't let a failure on one site abort processing of every
// other site queued after it in this run. Since we're not
// re-throwing, the outer wrapper's retry/requeue logic never
// sees this failure, so explicitly queue this resource for a
// follow-up pass to reconcile whatever this site didn't get to.
logger.error(
`rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} failed while processing site for siteResourceId=${siteResource.siteResourceId}, continuing with remaining sites and queuing a follow-up pass:`,
err
);
await rebuildQueue.enqueue({
type: "site-resource",
id: siteResource.siteResourceId
});
}
} }
// Handle subnet proxy target updates for the resource associations // Handle subnet proxy target updates for the resource associations
@@ -939,7 +960,7 @@ export async function updateClientSiteDestinations(
for (const site of sitesData) { for (const site of sitesData) {
if (!site.sites.subnet) { if (!site.sites.subnet) {
logger.warn(`Site ${site.sites.siteId} has no subnet, skipping`); logger.debug(`Site ${site.sites.siteId} has no subnet, skipping`);
continue; continue;
} }
@@ -1866,12 +1887,15 @@ async function rebuildClientAssociationsFromClientImpl(
// Insert new associations // Insert new associations
if (resourcesToAdd.length > 0) { if (resourcesToAdd.length > 0) {
await trx.insert(clientSiteResourcesAssociationsCache).values( await trx
resourcesToAdd.map((siteResourceId) => ({ .insert(clientSiteResourcesAssociationsCache)
clientId: client.clientId, .values(
siteResourceId resourcesToAdd.map((siteResourceId) => ({
})) clientId: client.clientId,
); siteResourceId
}))
)
.onConflictDoNothing();
} }
// Remove old associations // Remove old associations
@@ -1909,12 +1933,15 @@ async function rebuildClientAssociationsFromClientImpl(
// Insert new site associations // Insert new site associations
if (sitesToAdd.length > 0) { if (sitesToAdd.length > 0) {
await trx.insert(clientSitesAssociationsCache).values( await trx
sitesToAdd.map((siteId) => ({ .insert(clientSitesAssociationsCache)
clientId: client.clientId, .values(
siteId sitesToAdd.map((siteId) => ({
})) clientId: client.clientId,
); siteId
}))
)
.onConflictDoNothing();
} }
// Remove old site associations // Remove old site associations
server/routers/newt/buildConfiguration.ts
+2 -2
View File
@@ -52,13 +52,13 @@ export async function buildClientConfigurationForNewtClient(
clientsRes clientsRes
.filter((client) => { .filter((client) => {
if (!client.clients.pubKey) { if (!client.clients.pubKey) {
logger.warn( logger.debug(
`Client ${client.clients.clientId} has no public key, skipping` `Client ${client.clients.clientId} has no public key, skipping`
); );
return false; return false;
} }
if (!client.clients.subnet) { if (!client.clients.subnet) {
logger.warn( logger.debug(
`Client ${client.clients.clientId} has no subnet, skipping` `Client ${client.clients.clientId} has no subnet, skipping`
); );
return false; return false;
server/routers/olm/buildConfiguration.ts
+1 -1
View File
@@ -161,7 +161,7 @@ export async function buildSiteConfigurationForOlmClient(
} }
if (!site.subnet) { if (!site.subnet) {
logger.warn(`Site ${site.siteId} has no subnet, skipping`); logger.debug(`Site ${site.siteId} has no subnet, skipping`);
continue; continue;
} }