Provisioner asks the backend, not memory, whether a consumer is still there (hq issue 120) #7
+31
-12
@@ -114,6 +114,9 @@ export function runProvisioner(resource: string, adapter: Adapter, opts: Provisi
|
||||
}
|
||||
const h = hash(g.as, password, g.values ?? {});
|
||||
const p: Provision = { as: g.as, password, values: g.values ?? {}, at: g.at, consumer: g.node };
|
||||
// Set when the backend reported this consumer lost: how many successful re-applies in a row
|
||||
// it has now needed, counting this one.
|
||||
let reapplying: number | undefined;
|
||||
if (applied.get(g.as) === h) {
|
||||
if (!verifying) continue;
|
||||
const brake = lost.get(g.as);
|
||||
@@ -124,31 +127,44 @@ export function runProvisioner(resource: string, adapter: Adapter, opts: Provisi
|
||||
continue;
|
||||
}
|
||||
} catch (err) {
|
||||
// Unable to ask is not evidence of loss. Kept as applied, asked again next time. The rest
|
||||
// of this pass is not asked either: a backend that cannot answer for one consumer will not
|
||||
// answer for the next, and each would cost a timeout.
|
||||
// Unable to ask is not evidence of loss. Kept as applied, asked again next time.
|
||||
console.error(`[provisioner:${resource}] ${g.as}: could not check the backend, will ask again: ${scrub(err, password)}`);
|
||||
verifying = false;
|
||||
// A backend that did not answer in time will not answer for the next consumer either, and
|
||||
// each would cost a timeout, so the rest of this pass is not asked. Any other failure may
|
||||
// be this consumer's alone, and the others are still asked.
|
||||
if (err instanceof HoldsTimeout) verifying = false;
|
||||
continue;
|
||||
}
|
||||
const times = (brake?.times ?? 0) + 1;
|
||||
const waitMs = Math.min(MAX_BACKOFF_MS, verifyEveryMs * 2 ** (times - 1));
|
||||
lost.set(g.as, { times, nextAt: Date.now() + waitMs });
|
||||
if (times === 1) {
|
||||
reapplying = (brake?.times ?? 0) + 1;
|
||||
if (reapplying === 1) {
|
||||
// Said, because it means the backend lost something while nothing was looking.
|
||||
console.error(`[provisioner:${resource}] ${g.as}: the backend no longer holds it; applying again`);
|
||||
} else {
|
||||
console.error(
|
||||
`[provisioner:${resource}] ${g.as}: still not held after being applied again (${times} times in a row) — ` +
|
||||
`create does not produce what holds checks; applying again, next check in ${Math.round(waitMs / 1000)}s`,
|
||||
`[provisioner:${resource}] ${g.as}: still not held after being applied again (${reapplying} times in a row) — ` +
|
||||
`create does not produce what holds checks; applying again`,
|
||||
);
|
||||
}
|
||||
}
|
||||
try {
|
||||
await adapter.create(p);
|
||||
applied.set(g.as, h);
|
||||
if (reapplying === undefined) {
|
||||
// Applied for a new login, password or values: whatever was counted before does not carry over.
|
||||
lost.delete(g.as);
|
||||
} else {
|
||||
// Only a create that succeeded counts toward the brake: if the backend still does not hold
|
||||
// it at the next check, create and holds disagree, and each repeat waits twice as long.
|
||||
const waitMs = Math.min(MAX_BACKOFF_MS, verifyEveryMs * 2 ** (reapplying - 1));
|
||||
lost.set(g.as, { times: reapplying, nextAt: Date.now() + waitMs });
|
||||
if (reapplying > 1) {
|
||||
console.error(`[provisioner:${resource}] ${g.as}: next check in ${Math.round(waitMs / 1000)}s`);
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
console.error(`[provisioner:${resource}] ${g.as}: create failed, will retry: ${scrub(err, password)}`);
|
||||
// A failed create is not a disagreement: asked again at the next check, the count unchanged.
|
||||
if (reapplying !== undefined) lost.set(g.as, { times: reapplying - 1, nextAt: 0 });
|
||||
}
|
||||
}
|
||||
|
||||
@@ -210,11 +226,14 @@ async function readContributions(path: string, resource: string): Promise<Contri
|
||||
return (doc.given ?? []).filter((g) => g.as && g.secret);
|
||||
}
|
||||
|
||||
/** Reject with a timeout error if `p` has not settled within `ms`. */
|
||||
/** A check that did not answer in time: the backend, not the consumer, is the likely cause. */
|
||||
class HoldsTimeout extends Error {}
|
||||
|
||||
/** Reject with a HoldsTimeout if `p` has not settled within `ms`. */
|
||||
function withTimeout<T>(p: Promise<T>, ms: number): Promise<T> {
|
||||
let timer: NodeJS.Timeout | undefined;
|
||||
const timeout = new Promise<never>((_, reject) => {
|
||||
timer = setTimeout(() => reject(new Error(`no answer within ${ms}ms`)), ms);
|
||||
timer = setTimeout(() => reject(new HoldsTimeout(`no answer within ${ms}ms`)), ms);
|
||||
});
|
||||
return Promise.race([p, timeout]).finally(() => clearTimeout(timer));
|
||||
}
|
||||
|
||||
+67
-3
@@ -241,9 +241,9 @@ test("provisioner treats a failed or hung holds as could-not-ask, and never logs
|
||||
console.error = original;
|
||||
}
|
||||
assert.equal(creates, 2); // the first pass only; nothing is applied again on "could not ask"
|
||||
// After one consumer's check fails, the rest of that pass is not asked: "b" follows "a" and is
|
||||
// never reached, because every pass stops at "a".
|
||||
assert.ok(askedFor.length >= 2 && askedFor.every((as) => as === "a"), askedFor.join(","));
|
||||
// A failure that is not a timeout may be one consumer's alone: the consumers after it are still
|
||||
// asked on every pass.
|
||||
assert.ok(askedFor.filter((as) => as === "b").length >= 2, askedFor.join(","));
|
||||
assert.ok(logged.some((l) => l.includes("Command failed: tool --password ***")));
|
||||
assert.ok(!logged.some((l) => l.includes("s3cret-pw")));
|
||||
|
||||
@@ -272,6 +272,70 @@ test("provisioner treats a failed or hung holds as could-not-ask, and never logs
|
||||
assert.ok(hungLogged.some((l) => l.includes("no answer within 20ms")));
|
||||
});
|
||||
|
||||
test("provisioner stops a verify pass at a timeout, and retries a failed re-apply at the next check", async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "prov-retry-"));
|
||||
await writeFile(join(dir, "a.secret"), "pa");
|
||||
await writeFile(join(dir, "b.secret"), "pb");
|
||||
const receives = join(dir, "cache.json");
|
||||
await writeFile(
|
||||
receives,
|
||||
JSON.stringify({ requirement: "cache", given: [{ as: "a", secret: join(dir, "a.secret") }, { as: "b", secret: join(dir, "b.secret") }] }),
|
||||
);
|
||||
const original = console.error;
|
||||
console.error = () => {};
|
||||
try {
|
||||
// A timeout on "a" means "b" is not asked in that pass.
|
||||
const askedFor: string[] = [];
|
||||
const stopHung = runProvisioner(
|
||||
"cache",
|
||||
{
|
||||
async create() {},
|
||||
async remove() {},
|
||||
holds(p) {
|
||||
askedFor.push(p.as);
|
||||
return p.as === "a" ? new Promise<boolean>(() => {}) : Promise.resolve(true);
|
||||
},
|
||||
},
|
||||
{ receives, everyMs: 5, verifyEveryMs: 30, holdsTimeoutMs: 10 },
|
||||
);
|
||||
await new Promise((r) => setTimeout(r, 150));
|
||||
stopHung();
|
||||
assert.ok(askedFor.length >= 2 && askedFor.every((as) => as === "a"), askedFor.join(","));
|
||||
|
||||
// A lost consumer whose re-apply fails is retried at the next check, not held back by the brake,
|
||||
// and the retry that succeeds counts as the first re-apply, not the second.
|
||||
let held = true;
|
||||
let createCalls = 0;
|
||||
const logged: string[] = [];
|
||||
console.error = (...a: unknown[]) => logged.push(a.join(" "));
|
||||
const stop = runProvisioner(
|
||||
"cache",
|
||||
{
|
||||
async create(p) {
|
||||
if (p.as !== "a") return;
|
||||
createCalls++;
|
||||
if (createCalls === 2) throw new Error("write refused"); // the first re-apply fails
|
||||
if (createCalls === 3) held = true; // the retry works
|
||||
},
|
||||
async remove() {},
|
||||
async holds(p) {
|
||||
return p.as === "a" ? held : true;
|
||||
},
|
||||
},
|
||||
{ receives, everyMs: 5, verifyEveryMs: 20 },
|
||||
);
|
||||
await waitFor(() => createCalls === 1, 2000);
|
||||
held = false; // the backend loses "a"
|
||||
await waitFor(() => createCalls === 3, 2000);
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
stop();
|
||||
assert.equal(createCalls, 3);
|
||||
assert.ok(!logged.some((l) => l.includes("create does not produce what holds checks")), logged.join("\n"));
|
||||
} finally {
|
||||
console.error = original;
|
||||
}
|
||||
});
|
||||
|
||||
test("modules can SERVE: a real async tool, loaded and invoked over the broker", async () => {
|
||||
resetTools();
|
||||
|
||||
|
||||
Reference in New Issue
Block a user