Increase max attempt count for unprocessed items
This commit is contained in:
@@ -3209,7 +3209,10 @@ function saveUnprocessedSync(data: UnprocessedType): string {
|
|||||||
throw new Error('saveUnprocessedSync: id was falsey');
|
throw new Error('saveUnprocessedSync: id was falsey');
|
||||||
}
|
}
|
||||||
|
|
||||||
if (attempts >= MAX_UNPROCESSED_ATTEMPTS) {
|
if (attempts > MAX_UNPROCESSED_ATTEMPTS) {
|
||||||
|
logger.warn(
|
||||||
|
`saveUnprocessedSync: not saving ${id} due to exhausted attempts`
|
||||||
|
);
|
||||||
removeUnprocessedSync(id);
|
removeUnprocessedSync(id);
|
||||||
return id;
|
return id;
|
||||||
}
|
}
|
||||||
@@ -3376,7 +3379,7 @@ async function getAllUnprocessedAndIncrementAttempts(): Promise<
|
|||||||
.prepare<Query>(
|
.prepare<Query>(
|
||||||
`
|
`
|
||||||
DELETE FROM unprocessed
|
DELETE FROM unprocessed
|
||||||
WHERE attempts >= $MAX_UNPROCESSED_ATTEMPTS
|
WHERE attempts > $MAX_UNPROCESSED_ATTEMPTS
|
||||||
`
|
`
|
||||||
)
|
)
|
||||||
.run({ MAX_UNPROCESSED_ATTEMPTS });
|
.run({ MAX_UNPROCESSED_ATTEMPTS });
|
||||||
|
@@ -776,6 +776,13 @@ export default class MessageReceiver
|
|||||||
}
|
}
|
||||||
|
|
||||||
private async queueAllCached(): Promise<void> {
|
private async queueAllCached(): Promise<void> {
|
||||||
|
if (this.stoppingProcessing) {
|
||||||
|
log.info(
|
||||||
|
'MessageReceiver.queueAllCached: not running due to stopped processing'
|
||||||
|
);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
const items = await this.getAllFromCache();
|
const items = await this.getAllFromCache();
|
||||||
const max = items.length;
|
const max = items.length;
|
||||||
for (let i = 0; i < max; i += 1) {
|
for (let i = 0; i < max; i += 1) {
|
||||||
@@ -1098,7 +1105,7 @@ export default class MessageReceiver
|
|||||||
id,
|
id,
|
||||||
version: 2,
|
version: 2,
|
||||||
|
|
||||||
attempts: 1,
|
attempts: 0,
|
||||||
envelope: Bytes.toBase64(plaintext),
|
envelope: Bytes.toBase64(plaintext),
|
||||||
messageAgeSec: envelope.messageAgeSec,
|
messageAgeSec: envelope.messageAgeSec,
|
||||||
receivedAtCounter: envelope.receivedAtCounter,
|
receivedAtCounter: envelope.receivedAtCounter,
|
||||||
|
Reference in New Issue
Block a user