Stable twin of #1184. Both photo sweeps tracked whether they were running in a module-level variable, which is invisible to every other replica: a status poll routed to an idle replica reports isRunning false while another is mid-run, and the next POST starts a second pass over the whole library. Migration 179 adds one row per job, claimed with a conditional UPDATE whose affected-row count is the answer. The lease is fenced on a per-claim token so a runner superseded by a stale takeover cannot renew a claim it has lost or release one it no longer owns; renewal runs on a timer spanning the claim through release, since one hung NAS read can outlast the stale window inside a single iteration. maintenance_jobs is excluded from .picpeak archives. Gated on settings.edit / settings.view rather than main's system.manage, which does not exist on this branch — they are what settings.edit was later split into, so both branches let the same people through. Co-authored-by: Paul Nothaft <[email protected]>
This commit is contained in:
co-authored by
Paul Nothaft
parent
7f0ed23ea4
commit
58ccecc304
@@ -7,43 +7,106 @@ const fs = require('fs').promises;
|
||||
const logger = require('../utils/logger');
|
||||
|
||||
const { resolvePhotoFilePath } = require('../services/photoResolver');
|
||||
const maintenanceJobs = require('../services/maintenanceJobState');
|
||||
|
||||
// Module-level progress state
|
||||
let repairProgress = {
|
||||
isRunning: false,
|
||||
lastResult: null
|
||||
};
|
||||
// Run state for both sweeps lives in the database, not in this process
|
||||
// (#1181). It used to be a module-level object per job, which is correct on a
|
||||
// single replica and wrong behind a load balancer: the status poll answers
|
||||
// from whichever process it reaches, so an idle replica reports isRunning
|
||||
// false while another is mid-run, the UI re-enables the button, and the next
|
||||
// POST starts a duplicate pass over the entire library.
|
||||
//
|
||||
// Two separate rows, for the same reason the two objects were separate: the
|
||||
// jobs walk the same photos but read different things out of them, and one
|
||||
// running must not block or report for the other.
|
||||
const { JOB_DIMENSION_REPAIR, JOB_CAPTURE_DATE_BACKFILL } = maintenanceJobs;
|
||||
|
||||
// Same shape, separate state: the two jobs walk the same photos but read
|
||||
// different things out of them, and one running must not block or report for
|
||||
// the other.
|
||||
let captureDateProgress = {
|
||||
isRunning: false,
|
||||
lastResult: null
|
||||
};
|
||||
const { HEARTBEAT_INTERVAL_MS } = maintenanceJobs;
|
||||
|
||||
/**
|
||||
* Renew the lease on a timer for as long as the run holds it.
|
||||
*
|
||||
* On a timer, not between photos: a single hung read on a stalled NAS mount or
|
||||
* a slow S3 object can outlast the whole stale window inside one iteration, and
|
||||
* a renewal that only fires between photos never gets to run. The lease would
|
||||
* expire while the job was demonstrably alive, another replica would take it
|
||||
* over, and the two would walk the same rows — precisely the case the lease
|
||||
* exists to prevent. The timer also covers the candidate query, which on a
|
||||
* large library is itself slow.
|
||||
*
|
||||
* `lost()` reports whether the claim has since been taken over. The loops check
|
||||
* it between photos and stop: mid-photo interruption is not possible, so the
|
||||
* worst case is one extra row written by the old runner, and its release is
|
||||
* fenced on the token anyway.
|
||||
*/
|
||||
function startLeaseKeeper(jobName, token) {
|
||||
let lost = false;
|
||||
const timer = setInterval(async () => {
|
||||
try {
|
||||
if (!(await maintenanceJobs.heartbeat(jobName, token))) {
|
||||
lost = true;
|
||||
clearInterval(timer);
|
||||
}
|
||||
} catch (err) {
|
||||
// heartbeat() already swallows query errors and reports the claim as
|
||||
// held; this is belt-and-braces so an unexpected throw cannot kill the
|
||||
// timer callback and silently stop all renewals.
|
||||
logger.warn(`Lease renewal error for ${jobName}: ${err.message}`);
|
||||
}
|
||||
}, HEARTBEAT_INTERVAL_MS);
|
||||
// Do not hold the event loop open on account of a maintenance sweep.
|
||||
if (typeof timer.unref === 'function') timer.unref();
|
||||
|
||||
return {
|
||||
lost: () => lost,
|
||||
stop: () => clearInterval(timer),
|
||||
};
|
||||
}
|
||||
|
||||
// Repair photo dimensions (background job)
|
||||
router.post('/repair-dimensions', adminAuth, requirePermission('photos.edit'), async (req, res) => {
|
||||
try {
|
||||
if (repairProgress.isRunning) {
|
||||
// Claimed before the candidate query, not after: that query is an await,
|
||||
// and two requests arriving inside it would both read "not running" and
|
||||
// both start a pass. The claim is a conditional UPDATE, so it settles the
|
||||
// race across replicas as well as within one.
|
||||
const token = await maintenanceJobs.claim(JOB_DIMENSION_REPAIR);
|
||||
if (!token) {
|
||||
return res.status(409).json({ error: 'Repair is already running' });
|
||||
}
|
||||
// Started at the claim, not at the loop: on a large or loaded install the
|
||||
// candidate SELECT below (plus the setImmediate hop) can itself outlast the
|
||||
// stale window, and an unrenewed claim would be taken over before the sweep
|
||||
// had read its first photo. Handed to the background section, which stops
|
||||
// it; every early exit here stops it too.
|
||||
const lease = startLeaseKeeper(JOB_DIMENSION_REPAIR, token);
|
||||
|
||||
const photos = await db('photos')
|
||||
.join('events', 'photos.event_id', 'events.id')
|
||||
.where(function () {
|
||||
this.whereNull('photos.width').orWhereNull('photos.height');
|
||||
})
|
||||
.where(function () {
|
||||
this.where('photos.media_type', '!=', 'video').orWhereNull('photos.media_type');
|
||||
})
|
||||
.select(
|
||||
'photos.id', 'photos.path', 'photos.filename',
|
||||
'photos.source_origin', 'photos.external_relpath', 'photos.event_id',
|
||||
'events.source_mode', 'events.external_path', 'events.slug'
|
||||
);
|
||||
let photos;
|
||||
try {
|
||||
photos = await db('photos')
|
||||
.join('events', 'photos.event_id', 'events.id')
|
||||
.where(function () {
|
||||
this.whereNull('photos.width').orWhereNull('photos.height');
|
||||
})
|
||||
.where(function () {
|
||||
this.where('photos.media_type', '!=', 'video').orWhereNull('photos.media_type');
|
||||
})
|
||||
.select(
|
||||
'photos.id', 'photos.path', 'photos.filename',
|
||||
'photos.source_origin', 'photos.external_relpath', 'photos.event_id',
|
||||
'events.source_mode', 'events.external_path', 'events.slug'
|
||||
);
|
||||
} catch (err) {
|
||||
lease.stop();
|
||||
await maintenanceJobs.release(JOB_DIMENSION_REPAIR, token);
|
||||
throw err;
|
||||
}
|
||||
|
||||
if (photos.length === 0) {
|
||||
// Released with no result: nothing ran, so the numbers from the last
|
||||
// real run stay on screen rather than being blanked by a no-op.
|
||||
lease.stop();
|
||||
await maintenanceJobs.release(JOB_DIMENSION_REPAIR, token);
|
||||
return res.json({ message: 'No photos need dimension repair', count: 0 });
|
||||
}
|
||||
|
||||
@@ -54,70 +117,92 @@ router.post('/repair-dimensions', adminAuth, requirePermission('photos.edit'), a
|
||||
});
|
||||
|
||||
// Process in background
|
||||
repairProgress.isRunning = true;
|
||||
repairProgress.lastResult = null;
|
||||
|
||||
setImmediate(async () => {
|
||||
let sharp;
|
||||
try {
|
||||
sharp = require('sharp');
|
||||
} catch (err) {
|
||||
logger.error('Sharp not available for dimension repair:', err.message);
|
||||
repairProgress.isRunning = false;
|
||||
repairProgress.lastResult = { success: 0, failed: 0, error: 'Sharp not available' };
|
||||
lease.stop();
|
||||
await maintenanceJobs.release(JOB_DIMENSION_REPAIR, token, { success: 0, failed: 0, error: 'Sharp not available' });
|
||||
return;
|
||||
}
|
||||
|
||||
let successCount = 0;
|
||||
let errorCount = 0;
|
||||
let lostClaim = false;
|
||||
|
||||
for (const photo of photos) {
|
||||
try {
|
||||
const event = { source_mode: photo.source_mode, external_path: photo.external_path, slug: photo.slug };
|
||||
let fullPath;
|
||||
try {
|
||||
fullPath = resolvePhotoFilePath(event, photo);
|
||||
} catch (err) {
|
||||
logger.warn(`Photo ${photo.id} has no resolvable path, skipping dimension repair: ${err.message}`);
|
||||
errorCount++;
|
||||
continue;
|
||||
}
|
||||
// Everything below runs detached from the request, so an unexpected
|
||||
// throw has nobody to report to. Without this the claim would sit held
|
||||
// until it aged out of the staleness window, disabling the button for
|
||||
// that whole time on every replica.
|
||||
try {
|
||||
for (const photo of photos) {
|
||||
// The timer does the renewing; this only notices that it has
|
||||
// already failed, so the loop stops instead of running on beside the
|
||||
// replica that took the claim over.
|
||||
if (lease.lost()) { lostClaim = true; break; }
|
||||
|
||||
try {
|
||||
await fs.access(fullPath);
|
||||
} catch (err) {
|
||||
logger.warn(`File not found for photo ${photo.id}: ${fullPath}`);
|
||||
errorCount++;
|
||||
continue;
|
||||
}
|
||||
|
||||
const metadata = await sharp(fullPath).metadata();
|
||||
|
||||
if (metadata.width && metadata.height) {
|
||||
await db('photos')
|
||||
.where({ id: photo.id })
|
||||
.update({
|
||||
width: metadata.width,
|
||||
height: metadata.height
|
||||
});
|
||||
successCount++;
|
||||
|
||||
if (successCount % 50 === 0) {
|
||||
logger.info(`Dimension repair progress: ${successCount} updated...`);
|
||||
const event = { source_mode: photo.source_mode, external_path: photo.external_path, slug: photo.slug };
|
||||
let fullPath;
|
||||
try {
|
||||
fullPath = resolvePhotoFilePath(event, photo);
|
||||
} catch (err) {
|
||||
logger.warn(`Photo ${photo.id} has no resolvable path, skipping dimension repair: ${err.message}`);
|
||||
errorCount++;
|
||||
continue;
|
||||
}
|
||||
} else {
|
||||
logger.warn(`Could not extract dimensions for photo ${photo.id}`);
|
||||
|
||||
try {
|
||||
await fs.access(fullPath);
|
||||
} catch (err) {
|
||||
logger.warn(`File not found for photo ${photo.id}: ${fullPath}`);
|
||||
errorCount++;
|
||||
continue;
|
||||
}
|
||||
|
||||
const metadata = await sharp(fullPath).metadata();
|
||||
|
||||
if (metadata.width && metadata.height) {
|
||||
await db('photos')
|
||||
.where({ id: photo.id })
|
||||
.update({
|
||||
width: metadata.width,
|
||||
height: metadata.height
|
||||
});
|
||||
successCount++;
|
||||
|
||||
if (successCount % 50 === 0) {
|
||||
logger.info(`Dimension repair progress: ${successCount} updated...`);
|
||||
}
|
||||
} else {
|
||||
logger.warn(`Could not extract dimensions for photo ${photo.id}`);
|
||||
errorCount++;
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error(`Error repairing dimensions for photo ${photo.id}:`, error);
|
||||
errorCount++;
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error(`Error repairing dimensions for photo ${photo.id}:`, error);
|
||||
errorCount++;
|
||||
}
|
||||
}
|
||||
|
||||
repairProgress.isRunning = false;
|
||||
repairProgress.lastResult = { success: successCount, failed: errorCount };
|
||||
logger.info(`Dimension repair complete: ${successCount} success, ${errorCount} errors`);
|
||||
if (lostClaim) {
|
||||
// Another replica declared this run stale and took it over. It owns
|
||||
// the row now, so releasing would clear ITS flag — release() refuses
|
||||
// on the token, but there is nothing to report either way.
|
||||
logger.warn(`Dimension repair stopped: claim taken over after ${successCount} updated, ${errorCount} errors`);
|
||||
return;
|
||||
}
|
||||
await maintenanceJobs.release(JOB_DIMENSION_REPAIR, token, { success: successCount, failed: errorCount });
|
||||
logger.info(`Dimension repair complete: ${successCount} success, ${errorCount} errors`);
|
||||
} catch (err) {
|
||||
logger.error('Dimension repair aborted:', err);
|
||||
await maintenanceJobs
|
||||
.release(JOB_DIMENSION_REPAIR, token, { success: successCount, failed: errorCount, error: err.message })
|
||||
.catch(() => {});
|
||||
} finally {
|
||||
lease.stop();
|
||||
}
|
||||
});
|
||||
} catch (error) {
|
||||
logger.error('Error starting dimension repair:', error);
|
||||
@@ -146,13 +231,15 @@ router.get('/repair-dimensions/status', adminAuth, requirePermission('photos.vie
|
||||
|
||||
const total = Number(totalPhotos.count);
|
||||
const withDims = Number(withDimensions.count);
|
||||
// Read from the shared row, so this answers the same on every replica.
|
||||
const state = await maintenanceJobs.read(JOB_DIMENSION_REPAIR);
|
||||
|
||||
res.json({
|
||||
total,
|
||||
withDimensions: withDims,
|
||||
withoutDimensions: total - withDims,
|
||||
isRunning: repairProgress.isRunning,
|
||||
lastResult: repairProgress.lastResult
|
||||
isRunning: state.isRunning,
|
||||
lastResult: state.lastResult
|
||||
});
|
||||
} catch (error) {
|
||||
logger.error('Error fetching dimension repair status:', error);
|
||||
@@ -192,15 +279,18 @@ router.get('/repair-dimensions/status', adminAuth, requirePermission('photos.vie
|
||||
// the same people through.
|
||||
router.post('/repair-capture-dates', adminAuth, requirePermission('settings.edit'), async (req, res) => {
|
||||
try {
|
||||
if (captureDateProgress.isRunning) {
|
||||
// Claimed here, not after the candidate query: that query is an await, and
|
||||
// two POSTs arriving inside it would both read "not running" and both start
|
||||
// a pass over the same rows. Being a conditional UPDATE, the claim settles
|
||||
// that between replicas too. Every early exit below has to release it
|
||||
// again, hence the try/catch around the query.
|
||||
const token = await maintenanceJobs.claim(JOB_CAPTURE_DATE_BACKFILL);
|
||||
if (!token) {
|
||||
return res.status(409).json({ error: 'Capture date backfill is already running' });
|
||||
}
|
||||
// Claimed here, not after the candidate query: that query is an await, and
|
||||
// two POSTs arriving inside it would both read isRunning === false and both
|
||||
// start a pass over the same rows. Every early exit below has to release it
|
||||
// again, hence the try/catch around the query.
|
||||
captureDateProgress.isRunning = true;
|
||||
captureDateProgress.lastResult = null;
|
||||
// Started at the claim so the candidate SELECT below is covered too — see
|
||||
// the dimension repair above.
|
||||
const lease = startLeaseKeeper(JOB_CAPTURE_DATE_BACKFILL, token);
|
||||
|
||||
let photos;
|
||||
try {
|
||||
@@ -235,12 +325,15 @@ router.post('/repair-capture-dates', adminAuth, requirePermission('settings.edit
|
||||
'events.source_mode', 'events.external_path', 'events.slug'
|
||||
);
|
||||
} catch (err) {
|
||||
captureDateProgress.isRunning = false;
|
||||
lease.stop();
|
||||
await maintenanceJobs.release(JOB_CAPTURE_DATE_BACKFILL, token);
|
||||
throw err;
|
||||
}
|
||||
|
||||
if (photos.length === 0) {
|
||||
captureDateProgress.isRunning = false;
|
||||
// No result passed: nothing ran, so the last real run's numbers survive.
|
||||
lease.stop();
|
||||
await maintenanceJobs.release(JOB_CAPTURE_DATE_BACKFILL, token);
|
||||
return res.json({ message: 'No photos need a capture date', count: 0 });
|
||||
}
|
||||
|
||||
@@ -255,91 +348,109 @@ router.post('/repair-capture-dates', adminAuth, requirePermission('settings.edit
|
||||
let successCount = 0;
|
||||
let missingCount = 0;
|
||||
let errorCount = 0;
|
||||
let lostClaim = false;
|
||||
|
||||
for (const photo of photos) {
|
||||
try {
|
||||
const event = { source_mode: photo.source_mode, external_path: photo.external_path, slug: photo.slug };
|
||||
const isExternal = photo.source_origin === 'external' || photo.source_origin === 'reference';
|
||||
// Same reasoning as the dimension repair: detached from the request, so
|
||||
// an unexpected throw must not leave the claim held.
|
||||
try {
|
||||
for (const photo of photos) {
|
||||
// The timer renews; this only notices it has already failed.
|
||||
if (lease.lost()) { lostClaim = true; break; }
|
||||
|
||||
// Two source shapes, same split the thumbnail regenerator uses
|
||||
// (imageProcessor.js:391-420). External rows live on a local mount
|
||||
// and are read directly; managed rows live behind the storage
|
||||
// backend, which on an S3 install is not a filesystem at all — going
|
||||
// through resolvePhotoFilePath there would build a STORAGE_PATH that
|
||||
// holds nothing and fail every managed photo.
|
||||
let captured;
|
||||
if (isExternal) {
|
||||
let fullPath;
|
||||
try {
|
||||
fullPath = resolvePhotoFilePath(event, photo);
|
||||
} catch (err) {
|
||||
logger.warn(`Photo ${photo.id} has no resolvable path, skipping capture date: ${err.message}`);
|
||||
errorCount++;
|
||||
continue;
|
||||
try {
|
||||
const event = { source_mode: photo.source_mode, external_path: photo.external_path, slug: photo.slug };
|
||||
const isExternal = photo.source_origin === 'external' || photo.source_origin === 'reference';
|
||||
|
||||
// Two source shapes, same split the thumbnail regenerator uses
|
||||
// (imageProcessor.js:391-420). External rows live on a local mount
|
||||
// and are read directly; managed rows live behind the storage
|
||||
// backend, which on an S3 install is not a filesystem at all — going
|
||||
// through resolvePhotoFilePath there would build a STORAGE_PATH that
|
||||
// holds nothing and fail every managed photo.
|
||||
let captured;
|
||||
if (isExternal) {
|
||||
let fullPath;
|
||||
try {
|
||||
fullPath = resolvePhotoFilePath(event, photo);
|
||||
} catch (err) {
|
||||
logger.warn(`Photo ${photo.id} has no resolvable path, skipping capture date: ${err.message}`);
|
||||
errorCount++;
|
||||
continue;
|
||||
}
|
||||
|
||||
try {
|
||||
await fs.access(fullPath);
|
||||
} catch (err) {
|
||||
logger.warn(`File not found for photo ${photo.id}: ${fullPath}`);
|
||||
errorCount++;
|
||||
continue;
|
||||
}
|
||||
|
||||
captured = await extractCaptureDate(fullPath);
|
||||
} else {
|
||||
let sourceKey;
|
||||
try {
|
||||
sourceKey = resolvePhotoStorageKey(event, photo);
|
||||
} catch (err) {
|
||||
logger.warn(`Photo ${photo.id} has no resolvable storage key, skipping capture date: ${err.message}`);
|
||||
errorCount++;
|
||||
continue;
|
||||
}
|
||||
|
||||
// In local-fs mode withLocalCopy hands back the resolved path
|
||||
// without checking it exists, so the access probe stays. In S3
|
||||
// mode a missing object throws out of getToFile and lands in the
|
||||
// outer catch — both end up counted as failures, which is what a
|
||||
// missing original is.
|
||||
captured = await withLocalCopy(sourceKey, async (localPath) => {
|
||||
await fs.access(localPath);
|
||||
return extractCaptureDate(localPath);
|
||||
});
|
||||
}
|
||||
|
||||
try {
|
||||
await fs.access(fullPath);
|
||||
} catch (err) {
|
||||
logger.warn(`File not found for photo ${photo.id}: ${fullPath}`);
|
||||
errorCount++;
|
||||
continue;
|
||||
}
|
||||
|
||||
captured = await extractCaptureDate(fullPath);
|
||||
} else {
|
||||
let sourceKey;
|
||||
try {
|
||||
sourceKey = resolvePhotoStorageKey(event, photo);
|
||||
} catch (err) {
|
||||
logger.warn(`Photo ${photo.id} has no resolvable storage key, skipping capture date: ${err.message}`);
|
||||
errorCount++;
|
||||
continue;
|
||||
}
|
||||
|
||||
// In local-fs mode withLocalCopy hands back the resolved path
|
||||
// without checking it exists, so the access probe stays. In S3
|
||||
// mode a missing object throws out of getToFile and lands in the
|
||||
// outer catch — both end up counted as failures, which is what a
|
||||
// missing original is.
|
||||
captured = await withLocalCopy(sourceKey, async (localPath) => {
|
||||
await fs.access(localPath);
|
||||
return extractCaptureDate(localPath);
|
||||
});
|
||||
}
|
||||
|
||||
if (!captured) {
|
||||
if (!captured) {
|
||||
// No date recovered. Usually genuine — plenty of sources carry no
|
||||
// EXIF — but extractCaptureDate also returns null when the file is
|
||||
// unreadable as an image, so this bucket is "nothing to write",
|
||||
// not "definitely has no EXIF". The failure counter above is the
|
||||
// one that means the storage is broken.
|
||||
missingCount++;
|
||||
continue;
|
||||
}
|
||||
missingCount++;
|
||||
continue;
|
||||
}
|
||||
|
||||
// whereNull, not a blanket set: the job can run for a long time on a
|
||||
// large library, and an import or a replacement finishing meanwhile
|
||||
// has already written a date this pass would otherwise overwrite
|
||||
// with the same-or-worse value.
|
||||
const updated = await db('photos')
|
||||
.where({ id: photo.id })
|
||||
.whereNull('captured_at')
|
||||
.update({ captured_at: captured.toISOString() });
|
||||
if (updated) successCount++;
|
||||
// whereNull, not a blanket set: the job can run for a long time on a
|
||||
// large library, and an import or a replacement finishing meanwhile
|
||||
// has already written a date this pass would otherwise overwrite
|
||||
// with the same-or-worse value.
|
||||
const updated = await db('photos')
|
||||
.where({ id: photo.id })
|
||||
.whereNull('captured_at')
|
||||
.update({ captured_at: captured.toISOString() });
|
||||
if (updated) successCount++;
|
||||
|
||||
if (successCount % 50 === 0 && successCount > 0) {
|
||||
logger.info(`Capture date backfill progress: ${successCount} updated...`);
|
||||
if (successCount % 50 === 0 && successCount > 0) {
|
||||
logger.info(`Capture date backfill progress: ${successCount} updated...`);
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error(`Error backfilling capture date for photo ${photo.id}:`, error);
|
||||
errorCount++;
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error(`Error backfilling capture date for photo ${photo.id}:`, error);
|
||||
errorCount++;
|
||||
}
|
||||
}
|
||||
|
||||
captureDateProgress.isRunning = false;
|
||||
captureDateProgress.lastResult = { success: successCount, noExif: missingCount, failed: errorCount };
|
||||
logger.info(`Capture date backfill complete: ${successCount} updated, ${missingCount} without EXIF, ${errorCount} errors`);
|
||||
if (lostClaim) {
|
||||
logger.warn(`Capture date backfill stopped: claim taken over after ${successCount} updated, ${errorCount} errors`);
|
||||
return;
|
||||
}
|
||||
await maintenanceJobs.release(JOB_CAPTURE_DATE_BACKFILL, token, { success: successCount, noExif: missingCount, failed: errorCount });
|
||||
logger.info(`Capture date backfill complete: ${successCount} updated, ${missingCount} without EXIF, ${errorCount} errors`);
|
||||
} catch (err) {
|
||||
logger.error('Capture date backfill aborted:', err);
|
||||
await maintenanceJobs
|
||||
.release(JOB_CAPTURE_DATE_BACKFILL, token, { success: successCount, noExif: missingCount, failed: errorCount, error: err.message })
|
||||
.catch(() => {});
|
||||
} finally {
|
||||
lease.stop();
|
||||
}
|
||||
});
|
||||
} catch (error) {
|
||||
logger.error('Error starting capture date backfill:', error);
|
||||
@@ -383,13 +494,15 @@ router.get('/repair-capture-dates/status', adminAuth, requirePermission('setting
|
||||
|
||||
const total = Number(counts.total);
|
||||
const withCaptureDate = Number(counts.dated);
|
||||
// Read from the shared row, so this answers the same on every replica.
|
||||
const state = await maintenanceJobs.read(JOB_CAPTURE_DATE_BACKFILL);
|
||||
|
||||
res.json({
|
||||
total,
|
||||
withCaptureDate,
|
||||
withoutCaptureDate: total - withCaptureDate,
|
||||
isRunning: captureDateProgress.isRunning,
|
||||
lastResult: captureDateProgress.lastResult
|
||||
isRunning: state.isRunning,
|
||||
lastResult: state.lastResult
|
||||
});
|
||||
} catch (error) {
|
||||
logger.error('Error fetching capture date backfill status:', error);
|
||||
|
||||
@@ -0,0 +1,178 @@
|
||||
/**
|
||||
* Shared run state for the photo maintenance sweeps (#1181).
|
||||
*
|
||||
* These jobs used to keep `{ isRunning, lastResult }` in a module-level
|
||||
* variable. That is invisible to every other replica, so on a multi-replica
|
||||
* install the status endpoint answers from whichever process the poll happens
|
||||
* to reach and a second POST can start a duplicate pass over the whole
|
||||
* library. Moving the state into the database makes both the claim and the
|
||||
* reporting shared.
|
||||
*
|
||||
* The claim is a conditional UPDATE whose affected-row count is the answer —
|
||||
* the same shape backgroundProcessor uses to hand a photo to exactly one
|
||||
* worker (backgroundProcessor.js:110-116). Two replicas issuing it
|
||||
* concurrently cannot both match: the row is locked for the duration of each
|
||||
* UPDATE, so the loser sees is_running already true and gets 0 rows back.
|
||||
*
|
||||
* A lease that can be taken over needs fencing, which is what claim_token is
|
||||
* for. Taking over a stale claim does not stop the old runner — it is a
|
||||
* process nobody can signal, quite possibly still walking the library. So
|
||||
* every write it attempts carries the token it was issued:
|
||||
*
|
||||
* - heartbeat() reports whether the renewal landed. It returns false once
|
||||
* the claim has moved on, and the run loops treat that as "stop".
|
||||
* - release() only clears the row if the token still matches, so a
|
||||
* superseded runner finishing late cannot clear the new owner's flag or
|
||||
* overwrite its result.
|
||||
*
|
||||
* Without both of those, a takeover produces two live runners and the loser
|
||||
* ends up stomping the winner's state on its way out.
|
||||
*
|
||||
* Timestamps are written as ISO strings rather than Date objects. Production
|
||||
* stores Dates fine, but inside jest the sqlite3 binding turns them into the
|
||||
* literal string "[object Object]" (see CLAUDE.md), which would silently break
|
||||
* every staleness comparison in the tests. ISO-8601 also compares correctly
|
||||
* under SQLite's lexicographic text ordering, so the `<` below means the same
|
||||
* thing on both engines.
|
||||
*/
|
||||
|
||||
const os = require('os');
|
||||
const crypto = require('crypto');
|
||||
const { db } = require('../database/db');
|
||||
const logger = require('../utils/logger');
|
||||
|
||||
const JOB_DIMENSION_REPAIR = 'photo_dimension_repair';
|
||||
const JOB_CAPTURE_DATE_BACKFILL = 'photo_capture_date_backfill';
|
||||
|
||||
// How long a run may go without renewing its lease before another replica is
|
||||
// allowed to take it over. Generous on purpose: these jobs walk the whole
|
||||
// library and a single slow original on a stalled NAS mount can block the loop
|
||||
// for a while. The cost of being too eager is a duplicate pass; the cost of
|
||||
// being too patient is a button that stays disabled after a crash.
|
||||
const DEFAULT_STALE_MS = 15 * 60 * 1000;
|
||||
|
||||
// How often a running job renews. Time-based, and comfortably inside the stale
|
||||
// window: tying renewal to a photo counter meant a job whose photos were slow
|
||||
// — a stalled mount, a handful of very large originals — could be declared
|
||||
// abandoned while it was still working.
|
||||
const HEARTBEAT_INTERVAL_MS = 60 * 1000;
|
||||
|
||||
const OWNER = `${os.hostname()}:${process.pid}`;
|
||||
|
||||
const nowIso = () => new Date().toISOString();
|
||||
const cutoffIso = (staleAfterMs) => new Date(Date.now() - staleAfterMs).toISOString();
|
||||
|
||||
/**
|
||||
* Try to become the one runner of `jobName`.
|
||||
*
|
||||
* Returns a claim token on success, or null when another replica holds it and
|
||||
* is still renewing — the caller should answer 409. The token must be passed
|
||||
* to every subsequent heartbeat/release for this run.
|
||||
*/
|
||||
async function claim(jobName, { staleAfterMs = DEFAULT_STALE_MS } = {}) {
|
||||
const stamp = nowIso();
|
||||
const cutoff = cutoffIso(staleAfterMs);
|
||||
const token = crypto.randomBytes(16).toString('hex');
|
||||
|
||||
const claimed = await db('maintenance_jobs')
|
||||
.where({ job_name: jobName })
|
||||
.where(function () {
|
||||
// Free, or held by a run that has stopped renewing. heartbeat_at is
|
||||
// always written by the claim below, so a running job cannot have a null
|
||||
// heartbeat — no third case to handle here.
|
||||
this.where('is_running', false).orWhere('heartbeat_at', '<', cutoff);
|
||||
})
|
||||
.update({
|
||||
is_running: true,
|
||||
started_at: stamp,
|
||||
heartbeat_at: stamp,
|
||||
finished_at: null,
|
||||
owner: OWNER,
|
||||
claim_token: token,
|
||||
});
|
||||
|
||||
return claimed > 0 ? token : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Renew the lease.
|
||||
*
|
||||
* Returns false when this run no longer owns the claim — it was declared stale
|
||||
* and taken over. The caller must stop working at that point: the new owner is
|
||||
* already walking the same rows, and two runners writing is exactly what the
|
||||
* lock exists to prevent.
|
||||
*/
|
||||
async function heartbeat(jobName, token) {
|
||||
try {
|
||||
const renewed = await db('maintenance_jobs')
|
||||
.where({ job_name: jobName, is_running: true, claim_token: token })
|
||||
.update({ heartbeat_at: nowIso() });
|
||||
return renewed > 0;
|
||||
} catch (err) {
|
||||
// A failed renewal query is not proof the claim is gone, and aborting a
|
||||
// long sweep over one transient database blip is the worse trade. Say the
|
||||
// claim still holds; if it really has moved on, the next renewal says so.
|
||||
logger.warn(`maintenanceJobState: heartbeat failed for ${jobName}: ${err.message}`);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Give up the claim.
|
||||
*
|
||||
* Scoped to the token, so a runner that was superseded while it was working
|
||||
* cannot clear the new owner's flag or overwrite its result on the way out.
|
||||
* Returns false when the claim had already moved on.
|
||||
*
|
||||
* `result` is stored as the job's last outcome. Pass null (the "nothing to do"
|
||||
* and error paths) to release without overwriting what the previous real run
|
||||
* reported.
|
||||
*/
|
||||
async function release(jobName, token, result = null) {
|
||||
const update = { is_running: false, finished_at: nowIso() };
|
||||
if (result !== null && result !== undefined) {
|
||||
update.last_result = JSON.stringify(result);
|
||||
}
|
||||
const released = await db('maintenance_jobs')
|
||||
.where({ job_name: jobName, claim_token: token })
|
||||
.update(update);
|
||||
return released > 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Current state, in the shape the status endpoints hand to the frontend.
|
||||
*
|
||||
* A run whose lease has gone stale is reported as not running: the owning
|
||||
* replica is gone, nothing is going to release the claim, and the operator
|
||||
* needs the button back. The next claim() takes the row over on the same
|
||||
* condition, so the two agree.
|
||||
*/
|
||||
async function read(jobName, { staleAfterMs = DEFAULT_STALE_MS } = {}) {
|
||||
const row = await db('maintenance_jobs').where({ job_name: jobName }).first();
|
||||
if (!row) return { isRunning: false, lastResult: null };
|
||||
|
||||
const alive = row.heartbeat_at && new Date(row.heartbeat_at).getTime() > Date.now() - staleAfterMs;
|
||||
|
||||
let lastResult = null;
|
||||
if (row.last_result) {
|
||||
try {
|
||||
lastResult = JSON.parse(row.last_result);
|
||||
} catch (err) {
|
||||
// Never let a malformed row take the status endpoint down with it.
|
||||
logger.warn(`maintenanceJobState: unreadable last_result for ${jobName}: ${err.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
return { isRunning: Boolean(row.is_running) && Boolean(alive), lastResult };
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
claim,
|
||||
heartbeat,
|
||||
release,
|
||||
read,
|
||||
JOB_DIMENSION_REPAIR,
|
||||
JOB_CAPTURE_DATE_BACKFILL,
|
||||
DEFAULT_STALE_MS,
|
||||
HEARTBEAT_INTERVAL_MS,
|
||||
};
|
||||
@@ -29,7 +29,17 @@ const packageJson = require('../../package.json');
|
||||
const PICPEAK_FORMAT_VERSION = 1;
|
||||
|
||||
// Never exported as data — the target owns these (its own migrations set them).
|
||||
const EXCLUDED_TABLES = new Set(['knex_migrations', 'knex_migrations_lock']);
|
||||
const EXCLUDED_TABLES = new Set([
|
||||
'knex_migrations',
|
||||
'knex_migrations_lock',
|
||||
// Live lease state for the maintenance sweeps (#1181), not data. An archive
|
||||
// taken while a sweep was running would otherwise carry is_running = true and
|
||||
// a claim token belonging to a process on the source install. Restored within
|
||||
// the staleness window, the target reports the job as running and refuses new
|
||||
// POSTs, with no runner anywhere that could release it. The table is seeded
|
||||
// by its migration, so the target already has the rows it needs.
|
||||
'maintenance_jobs',
|
||||
]);
|
||||
|
||||
// Storage subdirs holding non-recalculable blobs — always included.
|
||||
const DOC_DIRS = ['business-docs', 'uploads'];
|
||||
|
||||
Reference in New Issue
Block a user