带租约和幂等控制的任务执行器

/*
 * 题目：带租约和幂等控制的任务执行器
 * 要求：实现内存任务仓库与 Worker：任务领取带租约、owner 和 version；
 * 租约过期后可重领，旧 Worker 不得提交晚到结果。相同幂等键只创建一次任务。
 * 示例：Worker A 租约过期后 B 重领，A 随后的 complete 必须被拒绝。
 */
class TaskRepository {
  constructor(now = Date.now) {}
  submit(payload, idempotencyKey) {}
  claim(owner, leaseMs) {}
  complete(id, owner, version, result) {}
  fail(id, owner, version, error, maxAttempts) {}
  assertLease(task, owner, version) {}
  get(id) {}
}
async function processOne(repository, workerId, handler, { leaseMs = 30000, maxAttempts = 3 } = {}) {}

(async () => {
  let now = 0;
  const repo = new TaskRepository(() => now);
  const first = repo.submit({ x: 1 }, "key");
  console.log(repo.submit({ x: 2 }, "key").id); // 预期：first.id
  const expired = repo.claim("expired-worker", 5);
  now = 5;
  console.log((() => { try { (() => repo.complete(first.id, "expired-worker", expired.version, 0))(); return false; } catch { return true; } })()); // 预期：true（抛出异常）
  const lease1 = repo.claim("w1", 10);
  now = 16;
  const lease2 = repo.claim("w2", 10);
  console.log(lease2.version > lease1.version); // 预期：true
})().catch(console.error);

// 关键机制：租约定义处理权，version 防止旧持有者提交；生产实现需事务性条件更新。
class TaskRepository {
  constructor(now = Date.now) {
    this.now = now;
    this.tasks = new Map();
    this.keys = new Map();
    this.sequence = 0;
  }
  submit(payload, idempotencyKey) {
    if (this.keys.has(idempotencyKey))
      return this.get(this.keys.get(idempotencyKey));
    const task = {
      id: `job${++this.sequence}`,
      idempotencyKey,
      payload,
      status: "pending",
      attempt: 0,
      version: 0,
      owner: null,
      leaseUntil: 0,
    };
    this.tasks.set(task.id, task);
    this.keys.set(idempotencyKey, task.id);
    return this.get(task.id);
  }
  claim(owner, leaseMs) {
    const now = this.now();
    const task = [...this.tasks.values()].find(
      (item) =>
        item.status === "pending" ||
        (item.status === "running" && item.leaseUntil <= now),
    );
    if (!task) return null;
    Object.assign(task, {
      status: "running",
      owner,
      leaseUntil: now + leaseMs,
      attempt: task.attempt + 1,
      version: task.version + 1,
    });
    return this.get(task.id);
  }
  complete(id, owner, version, result) {
    const task = this.tasks.get(id);
    this.assertLease(task, owner, version);
    Object.assign(task, {
      status: "completed",
      result,
      owner: null,
      leaseUntil: 0,
    });
    return this.get(id);
  }
  fail(id, owner, version, error, maxAttempts) {
    const task = this.tasks.get(id);
    this.assertLease(task, owner, version);
    Object.assign(task, {
      status: task.attempt >= maxAttempts ? "failed" : "pending",
      error: String(error?.message || error),
      owner: null,
      leaseUntil: 0,
    });
    return this.get(id);
  }
  assertLease(task, owner, version) {
    if (
      !task ||
      task.status !== "running" ||
      task.owner !== owner ||
      task.version !== version ||
      task.leaseUntil <= this.now()
    )
      throw new Error("stale or invalid lease");
  }
  get(id) {
    const value = this.tasks.get(id);
    return value ? structuredClone(value) : null;
  }
}
async function processOne(
  repository,
  workerId,
  handler,
  { leaseMs = 30000, maxAttempts = 3 } = {},
) {
  const task = repository.claim(workerId, leaseMs);
  if (!task) return null;
  try {
    const result = await handler(task.payload);
    return repository.complete(task.id, workerId, task.version, result);
  } catch (error) {
    return repository.fail(task.id, workerId, task.version, error, maxAttempts);
  }
}

// 三组以上可直接运行的验证；异步题使用独立执行块。
(async () => {
  let now = 0;
  const repo = new TaskRepository(() => now);
  const first = repo.submit({ x: 1 }, "key");
  console.log(repo.submit({ x: 2 }, "key").id); // 预期：first.id
  const expired = repo.claim("expired-worker", 5);
  now = 5;
  console.log((() => { try { (() => repo.complete(first.id, "expired-worker", expired.version, 0))(); return false; } catch { return true; } })()); // 预期：true（抛出异常）
  const lease1 = repo.claim("w1", 10);
  now = 16;
  const lease2 = repo.claim("w2", 10);
  console.log(lease2.version > lease1.version); // 预期：true
  console.log((() => { try { (() => repo.complete(first.id, "w1", lease1.version, 1))(); return false; } catch { return true; } })()); // 预期：true（抛出异常）
  repo.complete(first.id, "w2", lease2.version, 2);
  console.log(repo.get(first.id).result); // 预期：2
  const second = repo.submit({ n: 3 }, "key2");
  const done = await processOne(repo, "w", async ({ n }) => n * 2);
  console.log(done.id); // 预期：second.id
  console.log(done.result); // 预期：6


})().catch(console.error);