Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 | 1x 4x 3x 3x 1x 1x 1x 1x | // Part IV - Job state.
// In a real setup the job state (key: tags, value: status/progress/object name)
// would live in a database. Because the worker runs on the same instance as the
// API for this experiment, we just keep it in a static/global in-memory map so
// the producer, the worker and the status endpoint all share it.
//
// Each entry looks like:
// { status: 'queued' | 'processing' | 'done' | 'error',
// progress: 0..100,
// objectName: string | null,
// error: string | null }
const jobs = new Map();
function normalizeKey(tags) {
return String(tags || '').trim().toLowerCase();
}
function clampProgress(progress) {
const value = Number(progress);
if (Number.isNaN(value)) {
return 0;
}
return Math.max(0, Math.min(100, Math.round(value)));
}
function ensureEntry(key) {
let entry = jobs.get(key);
if (!entry) {
entry = { status: 'queued', progress: 0, objectName: null, error: null };
jobs.set(key, entry);
}
return entry;
}
// Mark a job as queued (freshly published, not picked up by the worker yet).
// Resets any previous state for these tags so re-zipping starts from scratch.
function markQueued(tags) {
jobs.set(normalizeKey(tags), {
status: 'queued',
progress: 0,
objectName: null,
error: null
});
}
// Update the coarse status of a job, optionally setting its progress.
function setStatus(tags, status, progress) {
const entry = ensureEntry(normalizeKey(tags));
entry.status = status;
if (progress !== undefined) {
entry.progress = clampProgress(progress);
}
}
// Update only the progress (0..100) of an in-flight job.
function setProgress(tags, progress) {
const entry = ensureEntry(normalizeKey(tags));
entry.progress = clampProgress(progress);
}
// Record that the zip for these tags is ready, storing the object name in GCS.
function markComplete(tags, objectName) {
jobs.set(normalizeKey(tags), {
status: 'done',
progress: 100,
objectName,
error: null
});
}
// Record that a job failed so the client can stop polling and show the error.
function markError(tags, message) {
const entry = ensureEntry(normalizeKey(tags));
entry.status = 'error';
entry.error = message || 'Unknown error';
}
// Returns the GCS object name for a finished job, or undefined if not ready yet.
function getObjectName(tags) {
const entry = jobs.get(normalizeKey(tags));
return entry && entry.status === 'done' ? entry.objectName : undefined;
}
// Returns the full status for these tags. When nothing is known yet we report
// an "idle" status so the client can tell "never queued" from "queued".
function getStatus(tags) {
const entry = jobs.get(normalizeKey(tags));
Eif (!entry) {
return { status: 'idle', progress: 0, objectName: null, error: null };
}
return {
status: entry.status,
progress: entry.progress,
objectName: entry.objectName,
error: entry.error
};
}
module.exports = {
markQueued,
setStatus,
setProgress,
markComplete,
markError,
getObjectName,
getStatus,
jobs
};
|