Terminplaner: Minutenjob in Verfall und Mailversand getrennt geschuetzt

fuehreAus nimmt { teil } und sperrt je job:teil; minutenJob startet Verfall
und Mailversand (Warteschlange plus Betriebspruefung) gemeinsam mit
Promise.all. Haengt ein Teil, laeuft der andere im naechsten Takt weiter
(verfallUebersprungen bzw. mailUebersprungen); uebersprungen gilt nur, wenn
beide Teile uebersprungen werden. Die Ueberlappungswarnung nennt den Teil,
das Fehlerlog behaelt minutenJob.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

server/services/termine/jobs.js

@@ -27,8 +27,10 @@ import { gibZellenFrei, raeumeVerwaisteZellen } from './zellen.js'
  * Hintergrund-Jobs des Terminplaners (Spezifikation 2026-10-02-terminplaner-design.md, Abschnitte 4 und 11). server/index.js
  * ruft sie mit new Date() auf; `now` ist hier immer ein Parameter.
  *
- *   minutenJob       jede Minute      abgelaufene Haltungen => verfallen (Zellen frei); Mail-Warteschlange senden; zuletzt die
- *                                     Betriebspruefung (Budget, Warteschlange, Anmeldepause => Log und Team-Mail, betrieb.js)
+ *   minutenJob       jede Minute      zwei Teile, gemeinsam gestartet (Promise.all) und je Teil gegen Ueberlappung geschuetzt:
+ *                                     Verfall: abgelaufene Haltungen => verfallen (Zellen frei);
+ *                                     Mailversand: Mail-Warteschlange senden, zuletzt die Betriebspruefung (Budget, Warteschlange,
+ *                                     Anmeldepause => Log und Team-Mail, betrieb.js)
  *   fuenfMinutenJob  alle 5 Minuten   Erinnerungen (G05); bestaetigt 1 Stunde nach Ende => erledigt; Anfragen mit Beginn in
  *                                     der Vergangenheit => abgelehnt (vom System, Zellen frei, keine Mail)
  *   taeglicherJob    taeglich         Loeschfristen (11), festgefahrene und alte Mails, abgeschaltete Funktionen, verwaiste Zellen;
@@ -38,7 +40,10 @@ import { gibZellenFrei, raeumeVerwaisteZellen } from './zellen.js'
  * Gemeinsame Regeln:
  * - Modus off: jeder Job kehrt sofort zurueck, ohne Abfrage, mit { aus: true }.
  * - Keine ueberlappenden Laeufe: laeuft ein Job noch, wird der naechste Takt uebersprungen ({ uebersprungen: true }, einmal im
- *   Log je Ueberlappung).
+ *   Log je Ueberlappung). Die Sperre gilt je `job:teil` (fuehreAus mit { teil }): haengt beim Minutenjob der Verfall, versendet der
+ *   naechste Takt trotzdem die Warteschlange und umgekehrt; uebersprungen gilt der Takt erst, wenn beide Teile uebersprungen sind
+ *   (sonst steht verfallUebersprungen bzw. mailUebersprungen im Ergebnis). Die Warnung nennt den Teil (`minutenJob (Verfall)`), das
+ *   Fehlerlog behaelt `minutenJob` als Jobnamen.
  * - Ein Schritt eines Jobs scheitert nie den ganzen Job: der Fehler steht im Log (Text des Fehlers, bei einer Buchung mit
  *   Buchungsnummer, nie mit Gastdaten), `fehler` im Ergebnis zaehlt ihn, die uebrigen Schritte laufen. Eine fehlende Collection
  *   (404 collection_not_found, Einrichtung unvollstaendig) meldet ein Job hoechstens einmal je Stunde.
@@ -100,23 +105,26 @@ async function schritt(lauf, was, fn, ersatz) {
 
 /**
  * Rahmen jedes Jobs: Schalter, Ueberlappung, Laufdaten. => Ergebnis von arbeit(lauf) | { aus: true } | { uebersprungen: true }
+ * `teil` (Name eines Teiljobs, z. B. „Verfall“) gibt dem Teil eine eigene Ueberlappungssperre `job:teil` und nennt ihn in der
+ * Warnung (`job (teil)`); `lauf.job` bleibt der Jobname (Fehlerlog, Warnung bei fehlender Collection).
  */
-async function fuehreAus(job, now, arbeit) {
+async function fuehreAus(job, now, arbeit, { teil = null } = {}) {
   const nowMs = pruefeNow(now, job)
   if (termineAus()) return { aus: true }
-  if (laufend.has(job)) {
-    if (!ueberlappGemeldet.has(job)) {
-      ueberlappGemeldet.add(job)
-      console.warn(`[termine-jobs] ${job}: der vorige Lauf ist noch nicht fertig, dieser Takt wird übersprungen`)
+  const sperre = teil === null ? job : `${job}:${teil}`
+  if (laufend.has(sperre)) {
+    if (!ueberlappGemeldet.has(sperre)) {
+      ueberlappGemeldet.add(sperre)
+      console.warn(`[termine-jobs] ${teil === null ? job : `${job} (${teil})`}: der vorige Lauf ist noch nicht fertig, dieser Takt wird übersprungen`)
     }
     return { uebersprungen: true }
   }
-  laufend.add(job)
+  laufend.add(sperre)
   try {
     return await arbeit({ job, nowMs, fehler: 0, gedeckelt: 0 })
   } finally {
-    laufend.delete(job)
-    ueberlappGemeldet.delete(job)
+    laufend.delete(sperre)
+    ueberlappGemeldet.delete(sperre)
   }
 }
 
@@ -214,19 +222,48 @@ async function verfalleHaltungen(lauf, now) {
 }
 
 /**
- * Jede Minute: abgelaufene Haltungen verfallen lassen, Warteschlange senden, zuletzt den Betrieb pruefen (Vorfaelle stehen im Log und
- * gehen als T20 ans Team; die Pruefung wirft nie und aendert das Ergebnis des Laufs nicht).
+ * Jede Minute, in zwei unabhaengig geschuetzten Teilen, die gemeinsam starten (Promise.all):
+ *  - Verfall: abgelaufene Haltungen verfallen lassen;
+ *  - Mailversand: Warteschlange senden, zuletzt den Betrieb pruefen (Vorfaelle stehen im Log und gehen als T20 ans Team; die Pruefung
+ *    wirft nie und aendert das Ergebnis des Laufs nicht).
+ * Haengt ein Teil, wird nur dieser im naechsten Takt uebersprungen (verfallUebersprungen bzw. mailUebersprungen im Ergebnis), der
+ * andere laeuft weiter. { uebersprungen: true } kommt erst, wenn beide Teile uebersprungen werden.
  * @param {Date} now
- * @returns {Promise<{ verfallen: number, faellig: number, gesendet: number, fehler: number } | { aus: true } | { uebersprungen: true }>}
+ * @returns {Promise<{ verfallen: number, faellig: number, gesendet: number, fehler: number, verfallUebersprungen?: true, mailUebersprungen?: true } | { aus: true } | { uebersprungen: true }>}
  */
 export async function minutenJob(now) {
-  return fuehreAus('minutenJob', now, async (lauf) => {
-    const verfallen = await schritt(lauf, 'Verfall', () => verfalleHaltungen(lauf, now), 0)
-    const { faellig, gesendet } = await schritt(lauf, 'Warteschlange', () => verarbeite(now), { faellig: 0, gesendet: 0 })
-    // Zuletzt: die Pruefung sieht den Stand nach Verfall und Versand (ein eben gesendeter Eintrag staut nichts mehr)
-    await schritt(lauf, 'Betriebsprüfung', () => pruefeBetrieb(now), [])
-    return { verfallen, faellig, gesendet, fehler: lauf.fehler }
-  })
+  const [verfall, mail] = await Promise.all([
+    fuehreAus(
+      'minutenJob',
+      now,
+      async (lauf) => {
+        const verfallen = await schritt(lauf, 'Verfall', () => verfalleHaltungen(lauf, now), 0)
+        return { verfallen, fehler: lauf.fehler }
+      },
+      { teil: 'Verfall' },
+    ),
+    fuehreAus(
+      'minutenJob',
+      now,
+      async (lauf) => {
+        const { faellig, gesendet } = await schritt(lauf, 'Warteschlange', () => verarbeite(now), { faellig: 0, gesendet: 0 })
+        // Zuletzt: die Pruefung sieht den Stand nach dem Versand (ein eben gesendeter Eintrag staut nichts mehr)
+        await schritt(lauf, 'Betriebsprüfung', () => pruefeBetrieb(now), [])
+        return { faellig, gesendet, fehler: lauf.fehler }
+      },
+      { teil: 'Mailversand' },
+    ),
+  ])
+  if (verfall.aus || mail.aus) return { aus: true }
+  if (verfall.uebersprungen && mail.uebersprungen) return { uebersprungen: true }
+  return {
+    verfallen: verfall.verfallen ?? 0,
+    faellig: mail.faellig ?? 0,
+    gesendet: mail.gesendet ?? 0,
+    fehler: (verfall.fehler ?? 0) + (mail.fehler ?? 0),
+    ...(verfall.uebersprungen ? { verfallUebersprungen: true } : {}),
+    ...(mail.uebersprungen ? { mailUebersprungen: true } : {}),
+  }
 }
 
 // ---------------------------------------------------------------- Fuenf-Minuten-Job

test/termineJobs.test.js

@@ -264,6 +264,54 @@ async function mitEinwurf(passt, dazwischen, fn) {
 }
 const listeVonBuchungen = (muster) => (url, methode) =>
   methode === 'GET' && /\/collections\/termineBookings\/documents(\?|$)/.test(url) && muster.test(url)
+const listeVonOutbox = (url, methode) => methode === 'GET' && /\/collections\/termineOutbox\/documents(\?|$)/.test(url)
+
+/**
+ * Haengende Schritte: haelt je Muster (url, methode) => boolean die erste Anfrage an, auf die es zutrifft, bis freigeben() alle
+ * loslaesst. warteBisRuhig() wartet, bis keine andere Anfrage mehr laeuft und 50 ms lang keine neue kam (die uebrigen Schritte
+ * des Takts sind dann durch). ende() stellt fetch wieder her.
+ */
+function mitHaengern(...muster) {
+  const echt = globalThis.fetch
+  const tore = muster.map(() => {
+    const tor = { gehalten: false }
+    tor.warte = new Promise((resolve) => (tor.los = resolve))
+    return tor
+  })
+  let offen = 0
+  let anfragen = 0
+  globalThis.fetch = async (url, optionen) => {
+    anfragen += 1
+    const text = decodeURIComponent(String(url))
+    const methode = optionen?.method ?? 'GET'
+    const i = muster.findIndex((passt, k) => !tore[k].gehalten && passt(text, methode))
+    if (i >= 0) {
+      tore[i].gehalten = true
+      await tore[i].warte
+      return echt(url, optionen)
+    }
+    offen += 1
+    try {
+      return await echt(url, optionen)
+    } finally {
+      offen -= 1
+    }
+  }
+  return {
+    freigeben: () => tore.forEach((tor) => tor.los()),
+    gehalten: () => tore.every((tor) => tor.gehalten),
+    async warteBisRuhig() {
+      for (let ruhig = 0; ruhig < 5; ) {
+        const stand = anfragen
+        await new Promise((resolve) => setTimeout(resolve, 10))
+        ruhig = offen === 0 && stand === anfragen ? ruhig + 1 : 0
+      }
+    },
+    ende: () => {
+      globalThis.fetch = echt
+    },
+  }
+}
 
 /** Alle Lese-Anfragen an die Collection scheitern wie in Appwrite mit 404 collection_not_found. */
 async function mitFehlenderCollection(collection, fn) {
@@ -1974,46 +2022,111 @@ test('fehlende Collection der Warteschlange: der Minutenjob verfällt trotzdem H
   assert.equal(warnLog().length, 1)
 })
 
-test('überlappende Läufe: ein langsamer Lauf wird vom nächsten Takt übersprungen (einmal im Log), danach geht es normal weiter', async () => {
+test('überlappende Läufe: hängen beide Teile des Minutenjobs, werden die nächsten Takte übersprungen (je Teil einmal im Log), danach geht es normal weiter', async () => {
   await legeBuchung('b_vorbei', { status: 'gehalten', heldUntil: iso(plus(JETZT, -MIN)) })
-  const echt = globalThis.fetch
-  let freigeben
-  const tor = new Promise((resolve) => (freigeben = resolve))
-  let erste = true
-  globalThis.fetch = async (url, optionen) => {
-    if (erste) {
-      erste = false
-      await tor
-    }
-    return echt(url, optionen)
-  }
+  const haenger = mitHaengern(listeVonBuchungen(/heldUntil/), listeVonOutbox)
   try {
     const langsam = minutenJob(JETZT)
+    await haenger.warteBisRuhig()
+    assert.ok(haenger.gehalten(), 'Verfall und Mailversand hängen')
     assert.deepEqual(await minutenJob(plus(JETZT, MIN)), { uebersprungen: true })
     assert.deepEqual(await minutenJob(plus(JETZT, 2 * MIN)), { uebersprungen: true })
-    assert.equal(warnLog().filter((zeile) => /minutenJob/.test(zeile)).length, 1, 'nur einmal gemeldet')
+    assert.equal(warnLog().filter((zeile) => /minutenJob \(Verfall\)/.test(zeile)).length, 1, 'Verfall nur einmal gemeldet')
+    assert.equal(warnLog().filter((zeile) => /minutenJob \(Mailversand\)/.test(zeile)).length, 1, 'Mailversand nur einmal gemeldet')
+    assert.equal(warnLog().filter((zeile) => /minutenJob/.test(zeile)).length, 2, 'sonst keine Warnung')
     // andere Jobs sind davon nicht betroffen
     assert.equal((await fuenfMinutenJob(JETZT)).uebersprungen, undefined)
-    freigeben()
+    haenger.freigeben()
     assert.equal((await langsam).verfallen, 1)
   } finally {
-    globalThis.fetch = echt
+    haenger.freigeben()
+    haenger.ende()
   }
-  assert.equal((await minutenJob(plus(JETZT, 3 * MIN))).uebersprungen, undefined, 'danach läuft der Job wieder')
+  assert.deepEqual(await minutenJob(plus(JETZT, 3 * MIN)), { verfallen: 0, faellig: 0, gesendet: 0, fehler: 0 }, 'danach laufen beide Teile wieder')
 })
 
-test('ein Job, der mit einem Fehler endet, gibt die Überlappungssperre frei', async () => {
+test('überlappende Läufe: hängt der Verfall, versendet der nächste Takt trotzdem die Warteschlange (Überlappung einmal im Log)', async () => {
+  await legeBuchung('b_vorbei', { status: 'gehalten', heldUntil: iso(plus(JETZT, -MIN)) })
+  legeEintrag('o_erst')
+  const haenger = mitHaengern(listeVonBuchungen(/heldUntil/))
+  try {
+    const langsam = minutenJob(JETZT)
+    await haenger.warteBisRuhig()
+    assert.ok(haenger.gehalten(), 'der Verfall hängt')
+    // beide Teile starten gemeinsam: die Warteschlange ist abgearbeitet, obwohl der Verfall des ersten Takts noch hängt
+    assert.equal(aw.docs('termineOutbox').get('o_erst').status, 'gesendet')
+    assert.equal(status('b_vorbei'), 'gehalten')
+
+    legeEintrag('o_zweit')
+    const zweiter = await minutenJob(plus(JETZT, MIN))
+    assert.deepEqual(zweiter, { verfallen: 0, faellig: 1, gesendet: 1, fehler: 0, verfallUebersprungen: true })
+    assert.equal(aw.docs('termineOutbox').get('o_zweit').status, 'gesendet')
+    assert.equal(sink.sent.length, 2)
+    const dritter = await minutenJob(plus(JETZT, 2 * MIN))
+    assert.equal(dritter.verfallUebersprungen, true)
+    assert.equal(dritter.uebersprungen, undefined, 'nur ein Teil übersprungen: der Takt gilt nicht als übersprungen')
+    assert.equal(warnLog().filter((zeile) => /minutenJob \(Verfall\)/.test(zeile)).length, 1, 'die Überlappung steht einmal im Log')
+    assert.equal(warnLog().filter((zeile) => /minutenJob/.test(zeile)).length, 1, 'der Mailversand wurde nicht übersprungen')
+
+    haenger.freigeben()
+    assert.deepEqual(await langsam, { verfallen: 1, faellig: 1, gesendet: 1, fehler: 0 })
+  } finally {
+    haenger.freigeben()
+    haenger.ende()
+  }
+  assert.equal(status('b_vorbei'), 'verfallen')
+  assert.deepEqual(await minutenJob(plus(JETZT, 3 * MIN)), { verfallen: 0, faellig: 0, gesendet: 0, fehler: 0 })
+})
+
+test('überlappende Läufe: hängt der Mailversand, verfällt im nächsten Takt eine abgelaufene Haltung trotzdem (Überlappung einmal im Log)', async () => {
+  const haenger = mitHaengern(listeVonOutbox)
+  try {
+    const langsam = minutenJob(JETZT)
+    await haenger.warteBisRuhig()
+    assert.ok(haenger.gehalten(), 'der Mailversand hängt')
+
+    await legeBuchung('b_vorbei', { status: 'gehalten', heldUntil: iso(plus(JETZT, MIN)) })
+    assert.ok(claims('b_vorbei').length > 0)
+    const zweiter = await minutenJob(plus(JETZT, 2 * MIN))
+    assert.deepEqual(zweiter, { verfallen: 1, faellig: 0, gesendet: 0, fehler: 0, mailUebersprungen: true })
+    assert.equal(status('b_vorbei'), 'verfallen')
+    assert.equal(claims('b_vorbei').length, 0, 'Zellen frei')
+    const dritter = await minutenJob(plus(JETZT, 3 * MIN))
+    assert.equal(dritter.mailUebersprungen, true)
+    assert.equal(dritter.uebersprungen, undefined)
+    assert.equal(warnLog().filter((zeile) => /minutenJob \(Mailversand\)/.test(zeile)).length, 1, 'die Überlappung steht einmal im Log')
+    assert.equal(warnLog().filter((zeile) => /minutenJob/.test(zeile)).length, 1, 'der Verfall wurde nicht übersprungen')
+
+    haenger.freigeben()
+    assert.deepEqual(await langsam, { verfallen: 0, faellig: 0, gesendet: 0, fehler: 0 })
+  } finally {
+    haenger.freigeben()
+    haenger.ende()
+  }
+  assert.deepEqual(await minutenJob(plus(JETZT, 4 * MIN)), { verfallen: 0, faellig: 0, gesendet: 0, fehler: 0 })
+})
+
+test('ein Job, der mit einem Fehler endet, gibt die Überlappungssperren frei', async () => {
   const echt = globalThis.fetch
   globalThis.fetch = async () => {
     throw new TypeError('fetch failed')
   }
   try {
     const erg = await minutenJob(JETZT)
-    assert.equal(erg.fehler, 2, 'Verfall und Warteschlange scheitern, der Job wirft nicht')
+    assert.equal(erg.fehler, 2, 'Verfall und Warteschlange scheitern (je Teil ein Fehler), der Job wirft nicht')
+    assert.deepEqual(
+      fehlerLog()
+        .map((zeile) => /^\[termine-jobs\] minutenJob: (Verfall|Warteschlange):/.exec(zeile)?.[1])
+        .filter(Boolean)
+        .sort(),
+      ['Verfall', 'Warteschlange'],
+      'das Fehlerlog behält minutenJob als Jobnamen',
+    )
   } finally {
     globalThis.fetch = echt
   }
-  assert.equal((await minutenJob(JETZT)).uebersprungen, undefined)
+  assert.deepEqual(await minutenJob(JETZT), { verfallen: 0, faellig: 0, gesendet: 0, fehler: 0 }, 'beide Sperren sind frei')
+  assert.equal(warnLog().filter((zeile) => /nicht fertig/.test(zeile)).length, 0)
 })
 
 // ---------------------------------------------------------------- Einbindung