Terminplaner: Mail-Warteschlange begrenzt und nach Vorrang geordnet

verarbeite liest nicht mehr alle wartenden Eintraege, sondern je Klasse
(vorrang, normal, nachrangig) eine begrenzte Liste (config.termine.mailListe,
Standard 100) mit select und orderAsc(naechsterVersuch). Bei Pause oder
leerem Gesamtbudget endet der Lauf ohne Abfrage, bei erreichter Reserve
entfaellt die nachrangige Klasse. Funktionen mit ausgeschoepftem Kontingent
kommen in eine Ausschlussliste (hoechstens 90 Werte); hoechstens vier weitere
Runden je Klasse, damit keine Funktion hinter einer vollen verhungert.

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

test/termineOutbox.test.js

@@ -69,6 +69,7 @@ beforeEach(async () => {
   config.termine.mailsProStunde = 200
   config.termine.mailsProFunktion = 40
   config.termine.mailsReserve = 0.25
+  config.termine.mailListe = 100
   tr = fakeTransport()
   setMailTransportForTests(tr)
   // Fehler-/Warnlogs der Warteschlange abfangen (Testausgabe bleibt sauber, Logs sind pruefbar)
@@ -80,6 +81,7 @@ afterEach(async () => {
   config.termine.mailsProStunde = 200
   config.termine.mailsProFunktion = 40
   config.termine.mailsReserve = 0.25
+  config.termine.mailListe = 100
   config.smtp.from = ABSENDER_VORHER
 })
 
@@ -220,7 +222,7 @@ test('sendeEintrag meldet das Ergebnis: gesendet, uebersprungen (unbekannt oder
   assert.equal(tr.aufrufe, 1)
 })
 
-test('verarbeite sendet nur fällige wartet-Einträge, älteste zuerst; gesendet und fehler bleiben unberührt', async () => {
+test('verarbeite sendet nur fällige wartet-Einträge, fälligste zuerst (naechsterVersuch, nicht createdAt); gesendet und fehler bleiben unberührt', async () => {
   const faellig = await legeAn({ bookingId: 'b_faellig', an: 'faellig@example.com', naechsterVersuch: plus(-5 * MIN).toISOString(), createdAt: plus(-9 * MIN).toISOString() })
   const aelter = await legeAn({ bookingId: 'b_aelter', an: 'aelter@example.com', naechsterVersuch: plus(-1 * MIN).toISOString(), createdAt: plus(-30 * MIN).toISOString() })
   const spaeter = await legeAn({ bookingId: 'b_spaeter', naechsterVersuch: plus(MIN).toISOString() })
@@ -229,7 +231,7 @@ test('verarbeite sendet nur fällige wartet-Einträge, älteste zuerst; gesendet
   const ergebnis = await verarbeite(JETZT)
   assert.deepEqual(ergebnis, { faellig: 2, gesendet: 2 })
   assert.equal(tr.aufrufe, 2)
-  assert.deepEqual(tr.senke.sent.map((m) => m.to[0].address), ['aelter@example.com', 'faellig@example.com'], 'älteste zuerst')
+  assert.deepEqual(tr.senke.sent.map((m) => m.to[0].address), ['faellig@example.com', 'aelter@example.com'], 'fälligste zuerst: naechsterVersuch -5 min vor -1 min, obwohl der andere Eintrag älter ist (createdAt)')
   assert.equal((await lies(faellig.$id)).status, 'gesendet')
   assert.equal((await lies(aelter.$id)).status, 'gesendet')
   assert.equal((await lies(spaeter.$id)).status, 'wartet')
@@ -238,7 +240,7 @@ test('verarbeite sendet nur fällige wartet-Einträge, älteste zuerst; gesendet
   await assert.rejects(() => verarbeite('jetzt'), TypeError)
 })
 
-test('verarbeite liest alle Seiten (mehr als 25 fällige Einträge)', async () => {
+test('verarbeite liest mit eigenem limit mehr als die 25 der Standardseite (30 fällige Einträge)', async () => {
   for (let i = 0; i < 30; i++) await legeAn({ bookingId: `b_${i}` })
   const ergebnis = await verarbeite(JETZT)
   assert.deepEqual(ergebnis, { faellig: 30, gesendet: 30 })
@@ -260,7 +262,7 @@ test('Stundenbudget 2: der dritte Eintrag wartet (unverändert, ohne Versuch) un
   assert.equal(wartend.versuche, 0)
   assert.ok(wartend.an && wartend.html, 'Inhalt bleibt erhalten')
 
-  assert.deepEqual(await verarbeite(plus(59 * MIN)), { faellig: 1, gesendet: 0 })
+  assert.deepEqual(await verarbeite(plus(59 * MIN)), { faellig: 0, gesendet: 0 }, 'Frühausstieg: das Budget ist voll, der wartende Eintrag wird nicht einmal gelesen')
   assert.equal(tr.senke.sent.length, 2, 'innerhalb der Stunde bleibt es bei zwei Mails')
   assert.deepEqual(await verarbeite(plus(61 * MIN)), { faellig: 1, gesendet: 1 })
   assert.equal(tr.senke.sent.length, 3)
@@ -337,10 +339,10 @@ test('Kontingent je Funktion: die Funktion A bekommt 3 von 5 Mails, B ist unber
   assert.equal(alle().filter((d) => d.status === 'wartet').length, 0)
 })
 
-test('verarbeite überspringt Einträge einer ausgeschöpften Funktion und bedient die Funktionen dahinter (älteste zuerst)', async () => {
+test('verarbeite überspringt Einträge einer ausgeschöpften Funktion und bedient die Funktionen dahinter (fälligste zuerst)', async () => {
   config.termine.mailsProFunktion = 2
-  for (let i = 0; i < 4; i++) await legeAn({ featureId: 'f_A', art: 'G02', bookingId: `b_a${i}`, createdAt: plus(i * 1000).toISOString() })
-  for (let i = 0; i < 2; i++) await legeAn({ featureId: 'f_B', art: 'G02', bookingId: `b_b${i}`, createdAt: plus(10000 + i * 1000).toISOString() })
+  for (let i = 0; i < 4; i++) await legeAn({ featureId: 'f_A', art: 'G02', bookingId: `b_a${i}`, naechsterVersuch: plus(-60000 + i * 1000).toISOString() })
+  for (let i = 0; i < 2; i++) await legeAn({ featureId: 'f_B', art: 'G02', bookingId: `b_b${i}`, naechsterVersuch: plus(-50000 + i * 1000).toISOString() })
   assert.deepEqual(await verarbeite(plus(MIN)), { faellig: 6, gesendet: 4 })
   const gesendet = alle().filter((d) => d.status === 'gesendet')
   assert.deepEqual(gesendet.map((d) => d.bookingId).sort(), ['b_a0', 'b_a1', 'b_b0', 'b_b1'])
@@ -360,7 +362,7 @@ test('Vorrang-Reserve: nachrangige Mails (G05, K20, K21) nehmen höchstens 6 von
   await warteAufVersand()
   assert.equal(tr.senke.sent.length, 7, 'Platz 7 geht an den Vorrang')
 
-  assert.deepEqual(await verarbeite(plus(2 * MIN)), { faellig: 4, gesendet: 0 }, 'das siebte G05 wartet, obwohl noch ein Platz frei ist')
+  assert.deepEqual(await verarbeite(plus(2 * MIN)), { faellig: 0, gesendet: 0 }, 'das siebte G05 wartet, obwohl noch ein Platz frei ist (Reserve erreicht: die nachrangigen werden nicht einmal gelesen)')
   assert.equal(tr.senke.sent.length, 7)
   assert.ok(alle().filter((d) => d.status === 'wartet').every((d) => d.versuche === 0 && d.html))
 
@@ -499,6 +501,176 @@ test('Reserve: mit sehr kleinem Stundenbudget darf mindestens eine nachrangige M
   assert.equal(mailBudget(JETZT).nachrangig.max, 1, 'floor(3 × 0.1) = 0 ⇒ mindestens 1')
 })
 
+// ---------------------------------------------------------------------------------------------------------------------
+// Minutenjob: begrenzte, nach Klasse geordnete Liste (verarbeite)
+// ---------------------------------------------------------------------------------------------------------------------
+
+const listenAbfragen = () => aw.listCalls('termineOutbox')
+const abfrageKlasse = ({ queries }) => {
+  const art = queries.find((q) => q.attribute === 'art')
+  if (art.method === 'notEqual') return 'normal'
+  return art.values.includes('G01') ? 'vorrang' : 'nachrangig'
+}
+const abfrageKlassen = () => listenAbfragen().map(abfrageKlasse)
+const ausschluss = ({ queries }) => queries.find((q) => q.method === 'notEqual' && q.attribute === 'featureId')?.values ?? null
+const begrenzteLeseabfragen = (abfragen) => {
+  for (const abfrage of abfragen) {
+    const limit = abfrage.queries.find((q) => q.method === 'limit')
+    assert.ok(limit && limit.values[0] <= 100, `limit höchstens 100: ${JSON.stringify(abfrage.queries)}`)
+    assert.ok(!abfrage.queries.some((q) => q.method === 'cursorAfter' || (q.method === 'orderAsc' && q.attribute === '$id')), 'kein $id-Seitenlauf')
+    assert.deepEqual(abfrage.queries.find((q) => q.method === 'select')?.values, ['$id', 'featureId', 'art', 'naechsterVersuch'], 'nur die vier Felder')
+    assert.ok(abfrage.queries.some((q) => q.method === 'orderAsc' && q.attribute === 'naechsterVersuch'), 'fälligste zuerst')
+    assert.ok(!JSON.stringify(abfrage.queries).includes('html'), 'kein Mailtext in der Abfrage')
+    assert.ok(abfrage.queries.some((q) => q.method === 'equal' && q.attribute === 'status' && q.values[0] === 'wartet'))
+    assert.ok(abfrage.queries.some((q) => q.method === 'lessThanEqual' && q.attribute === 'naechsterVersuch'))
+  }
+}
+/** Alle Einträge löschen und den Prozesszustand zurücksetzen (mehrere Szenarien in einem Test). */
+async function leereWarteschlange() {
+  for (const id of aw.docs('termineOutbox').keys()) await deleteDocument(col(), id)
+  resetOutboxForTests()
+  tr = fakeTransport()
+  setMailTransportForTests(tr)
+}
+const empfaenger = () => tr.senke.sent.map((m) => m.to[0].address)
+
+test('Frühausstieg: bei leerem Gesamtbudget liest verarbeite nichts (keine Listenabfrage an termineOutbox)', async () => {
+  config.termine.mailsProStunde = 2
+  for (const nummer of [1, 2]) await einreihen(eintrag({ bookingId: `b_${nummer}` }), JETZT)
+  await warteAufVersand()
+  await legeAn({ art: 'G02', bookingId: 'b_wartet', naechsterVersuch: plus(-MIN).toISOString() })
+  aw.resetListCalls()
+  assert.deepEqual(await verarbeite(plus(MIN)), { faellig: 0, gesendet: 0 })
+  assert.deepEqual(await verarbeite(plus(59 * MIN)), { faellig: 0, gesendet: 0 })
+  assert.equal(listenAbfragen().length, 0, 'Budget voll: nicht eine Abfrage')
+  assert.equal(tr.senke.sent.length, 2)
+
+  assert.deepEqual(await verarbeite(plus(61 * MIN)), { faellig: 1, gesendet: 1 }, 'nach einer Stunde wird gelesen und gesendet')
+  assert.ok(listenAbfragen().length > 0)
+})
+
+test('Frühausstieg: während der Anmeldepause liest verarbeite nichts, danach läuft es weiter', async () => {
+  tr.fehler.push(EAUTH)
+  await einreihen(eintrag(), JETZT) // Anmeldefehler: 15 Minuten Pause, der Eintrag bleibt wartet
+  await warteAufVersand()
+  assert.equal(pausiertBis().getTime(), plus(15 * MIN).getTime())
+  await legeAn({ art: 'G02', bookingId: 'b_wartet' })
+  aw.resetListCalls()
+  assert.deepEqual(await verarbeite(plus(MIN)), { faellig: 0, gesendet: 0 })
+  assert.equal(listenAbfragen().length, 0, 'Pause: nicht eine Abfrage')
+  assert.deepEqual(await verarbeite(plus(16 * MIN)), { faellig: 2, gesendet: 2 })
+})
+
+test('begrenzte Liste: 250 fällige G02 aus 3 Funktionen ⇒ jede Abfrage limit ≤ 100, ohne $id-Seitenlauf, mit select; gesendet wird bis zum Kontingent', async () => {
+  const funktionen = ['f_A', 'f_B', 'f_C']
+  for (let i = 0; i < 250; i++) {
+    await legeAn({ featureId: funktionen[i % 3], art: 'G02', bookingId: `b_${i}`, naechsterVersuch: plus(-3600000 + i * 1000).toISOString() })
+  }
+  aw.resetListCalls()
+  const ergebnis = await verarbeite(JETZT)
+  assert.deepEqual(ergebnis, { faellig: 200, gesendet: 120 }, 'zwei volle Seiten gelesen; je Funktion höchstens 40 (Kontingent)')
+  begrenzteLeseabfragen(listenAbfragen())
+  assert.deepEqual(abfrageKlassen(), ['vorrang', 'vorrang', 'vorrang', 'normal', 'nachrangig'])
+  for (const f of funktionen) assert.equal(alle().filter((d) => d.featureId === f && d.status === 'gesendet').length, 40)
+  assert.equal(alle().filter((d) => d.status === 'wartet').length, 130)
+  assert.ok(alle().filter((d) => d.status === 'wartet').every((d) => d.versuche === 0 && d.html), 'gebremste Einträge bleiben unverändert')
+  assert.equal(tr.aufrufe, 120)
+})
+
+test('Reihenfolge: Vorrang vor normal vor nachrangig, innerhalb einer Klasse die fälligste zuerst (Budget 1 bis 4)', async () => {
+  const lege = async () => {
+    await legeAn({ art: 'G05', an: 'g05@example.com', bookingId: 'b_g05', naechsterVersuch: plus(-10 * MIN).toISOString(), createdAt: plus(-10 * MIN).toISOString() })
+    await legeAn({ art: 'G03', an: 'g03a@example.com', bookingId: 'b_g03a', naechsterVersuch: plus(-3 * MIN).toISOString(), createdAt: plus(-60 * MIN).toISOString() })
+    await legeAn({ art: 'G03', an: 'g03b@example.com', bookingId: 'b_g03b', naechsterVersuch: plus(-7 * MIN).toISOString(), createdAt: plus(-MIN).toISOString() })
+    await legeAn({ art: 'G01', an: 'g01@example.com', bookingId: 'b_g01', naechsterVersuch: plus(-MIN).toISOString(), createdAt: plus(-30 * MIN).toISOString() })
+  }
+  // gelesen wird je Klasse eine Liste: vorrang 1 Eintrag, normal 2, nachrangig 1; danach ist das Budget voll oder alles gesendet
+  const szenarien = [
+    [1, ['g01@example.com'], 1],
+    [2, ['g01@example.com', 'g03b@example.com'], 3],
+    [3, ['g01@example.com', 'g03b@example.com', 'g03a@example.com'], 3],
+    [4, ['g01@example.com', 'g03b@example.com', 'g03a@example.com', 'g05@example.com'], 4],
+  ]
+  for (const [budget, erwartet, gelesen] of szenarien) {
+    await leereWarteschlange()
+    config.termine.mailsProStunde = budget
+    await lege()
+    const ergebnis = await verarbeite(JETZT)
+    assert.deepEqual(empfaenger(), erwartet, `Budget ${budget}`)
+    assert.equal(ergebnis.gesendet, budget)
+    assert.equal(ergebnis.faellig, gelesen, `Budget ${budget}: gelesen wird nur, bis das Budget voll ist`)
+    assert.equal(alle().filter((d) => d.status === 'wartet').length, 4 - budget)
+  }
+})
+
+test('kein Verhungern: 150 fällige G02 der Funktion A (Kontingent 40) vor einem G02 der Funktion B ⇒ ein Lauf sendet 40 von A und die Mail von B', async () => {
+  for (let i = 0; i < 150; i++) await legeAn({ featureId: 'f_A', art: 'G02', bookingId: `b_a${i}`, naechsterVersuch: plus(-3600000 + i * 1000).toISOString() })
+  const b = await legeAn({ featureId: 'f_B', art: 'G02', bookingId: 'b_b', an: 'b@example.com', naechsterVersuch: plus(-1000).toISOString() })
+  aw.resetListCalls()
+  assert.deepEqual(await verarbeite(JETZT), { faellig: 101, gesendet: 41 })
+  assert.equal((await lies(b.$id)).status, 'gesendet')
+  assert.equal(alle().filter((d) => d.featureId === 'f_A' && d.status === 'gesendet').length, 40)
+  const abfragen = listenAbfragen()
+  begrenzteLeseabfragen(abfragen)
+  const vorrang = abfragen.filter((a) => abfrageKlasse(a) === 'vorrang')
+  assert.equal(vorrang.length, 2, 'erste Seite voll von A, zweite mit Ausschlussliste')
+  assert.equal(ausschluss(vorrang[0]), null, 'die erste Abfrage schließt nichts aus')
+  assert.deepEqual(ausschluss(vorrang[1]), ['f_A'])
+  for (const klasse of ['vorrang', 'normal', 'nachrangig']) assert.ok(abfragen.filter((a) => abfrageKlasse(a) === klasse).length <= 5)
+})
+
+test('Runden: höchstens 5 Abfragen je Klasse; mailListe wird bei jedem Aufruf gelesen; der Rest folgt beim nächsten Lauf', async () => {
+  config.termine.mailListe = 2
+  for (let i = 0; i < 12; i++) await legeAn({ featureId: `f_${i}`, art: 'G02', bookingId: `b_${i}`, naechsterVersuch: plus(-3600000 + i * 1000).toISOString() })
+  aw.resetListCalls()
+  assert.deepEqual(await verarbeite(JETZT), { faellig: 10, gesendet: 10 })
+  assert.deepEqual(abfrageKlassen(), ['vorrang', 'vorrang', 'vorrang', 'vorrang', 'vorrang', 'normal', 'nachrangig'])
+  assert.ok(listenAbfragen().every((a) => a.queries.find((q) => q.method === 'limit').values[0] === 2))
+  assert.equal(alle().filter((d) => d.status === 'wartet').length, 2)
+  assert.deepEqual(await verarbeite(JETZT), { faellig: 2, gesendet: 2 })
+})
+
+test('Ausschlussliste: höchstens 90 Funktionen je Abfrage, auch wenn mehr blockiert sind', async () => {
+  config.termine.mailsProFunktion = 1
+  for (let i = 0; i < 100; i++) {
+    for (const teil of ['a', 'b']) {
+      await legeAn({ featureId: `f_${i}`, art: 'G02', bookingId: `b_${i}${teil}`, naechsterVersuch: plus(-3600000 + (2 * i + (teil === 'a' ? 0 : 1)) * 1000).toISOString() })
+    }
+  }
+  aw.resetListCalls()
+  assert.deepEqual(await verarbeite(JETZT), { faellig: 210, gesendet: 100 })
+  const abfragen = listenAbfragen()
+  const vorrang = abfragen.filter((a) => abfrageKlasse(a) === 'vorrang')
+  assert.deepEqual(vorrang.map((a) => ausschluss(a)?.length ?? 0), [0, 50, 90], 'Runde 3: 100 blockierte Funktionen, 90 davon in der Liste')
+  for (const abfrage of abfragen) assert.ok((ausschluss(abfrage)?.length ?? 0) <= 90)
+  assert.ok(abfragen.filter((a) => abfrageKlasse(a) !== 'vorrang').every((a) => ausschluss(a).length === 90), 'blockierte Funktionen gelten für alle Klassen')
+  assert.equal(alle().filter((d) => d.status === 'gesendet').length, 100)
+})
+
+test('Reserve erreicht: keine Abfrage der nachrangigen Klasse mehr; Vorrang und normal kommen trotzdem dran', async () => {
+  config.termine.mailsProStunde = 8 // nachrangig höchstens 6
+  for (let i = 1; i <= 6; i++) await einreihen(von('f_A', 'G05', i), JETZT)
+  await warteAufVersand()
+  assert.equal(tr.senke.sent.length, 6)
+  const rest = await legeAn({ featureId: 'f_A', art: 'G05', bookingId: 'b_rest', naechsterVersuch: plus(-MIN).toISOString() })
+  await legeAn({ featureId: 'f_A', art: 'G03', bookingId: 'b_normal', naechsterVersuch: plus(-MIN).toISOString() })
+  aw.resetListCalls()
+  assert.deepEqual(await verarbeite(plus(MIN)), { faellig: 1, gesendet: 1 })
+  assert.deepEqual(abfrageKlassen(), ['vorrang', 'normal'], 'die nachrangige Klasse wird gar nicht erst gelesen')
+  assert.equal((await lies(rest.$id)).status, 'wartet')
+  assert.equal((await lies(rest.$id)).versuche, 0)
+})
+
+test('Reserve mitten im Lauf: nach der Runde, die sie erreicht, folgt keine weitere nachrangige Abfrage', async () => {
+  config.termine.mailsProStunde = 8 // nachrangig höchstens 6
+  config.termine.mailListe = 3
+  for (let i = 0; i < 10; i++) await legeAn({ art: i % 2 ? 'K20' : 'G05', bookingId: `b_${i}`, naechsterVersuch: plus(-3600000 + i * 1000).toISOString() })
+  aw.resetListCalls()
+  assert.deepEqual(await verarbeite(JETZT), { faellig: 6, gesendet: 6 })
+  assert.deepEqual(abfrageKlassen(), ['vorrang', 'normal', 'nachrangig', 'nachrangig'], 'zwei Runden à 3; vor der dritten ist die Reserve erreicht')
+  assert.equal(alle().filter((d) => d.status === 'wartet').length, 4)
+})
+
 // ---------------------------------------------------------------------------------------------------------------------
 // Fehlerklassen
 // ---------------------------------------------------------------------------------------------------------------------

test/termineSchalter.test.js

@@ -49,6 +49,7 @@ test('Konfiguration: Vorgaben ohne Umgebungsvariablen', async () => {
     mailsProStunde: 200,
     mailsProFunktion: 40,
     mailsReserve: 0.25,
+    mailListe: 100,
     loeschDeckel: 2000,
     loeschDeckelGesamt: 20000,
     zellenDeckel: 5000,