Fix handling of encrypted unprocessed envelopes

This commit is contained in:
Fedor Indutny 2025-02-10 12:44:59 -08:00 committed by GitHub
parent 0d87e3e6c9
commit 5bdb39a95b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 73 additions and 23 deletions

View file

@ -375,6 +375,11 @@ export type StickerPackRefType = Readonly<{
export type UnprocessedType = {
id: string;
timestamp: number;
/*
* A client generated date used for removing old envelopes from the table
* on startup.
*/
receivedAtDate: number;
receivedAtCounter: number;
attempts: number;
type: number;

View file

@ -4629,6 +4629,7 @@ function saveUnprocessed(db: WritableDB, data: UnprocessedType): string {
const {
id,
timestamp,
receivedAtDate,
receivedAtCounter,
attempts,
type,
@ -4659,6 +4660,7 @@ function saveUnprocessed(db: WritableDB, data: UnprocessedType): string {
id,
timestamp,
receivedAtCounter,
receivedAtDate,
attempts,
type,
isEncrypted,
@ -4680,6 +4682,7 @@ function saveUnprocessed(db: WritableDB, data: UnprocessedType): string {
$id,
$timestamp,
$receivedAtCounter,
$receivedAtDate,
$attempts,
$type,
$isEncrypted,
@ -4702,7 +4705,8 @@ function saveUnprocessed(db: WritableDB, data: UnprocessedType): string {
).run({
id,
timestamp,
receivedAtCounter: receivedAtCounter ?? null,
receivedAtCounter,
receivedAtDate,
attempts,
type,
isEncrypted: isEncrypted ? 1 : 0,
@ -4750,9 +4754,11 @@ function getAllUnprocessedIds(db: WritableDB): Array<string> {
return db.transaction(() => {
// cleanup first
const { changes: deletedStaleCount } = db
.prepare<Query>('DELETE FROM unprocessed WHERE timestamp < $monthAgo')
.prepare<Query>(
'DELETE FROM unprocessed WHERE receivedAtDate < $messageQueueCutoff'
)
.run({
monthAgo: Date.now() - 45 * durations.DAY,
messageQueueCutoff: Date.now() - 45 * durations.DAY,
});
if (deletedStaleCount !== 0) {

View file

@ -0,0 +1,38 @@
// Copyright 2025 Signal Messenger, LLC
// SPDX-License-Identifier: AGPL-3.0-only
import type { LoggerType } from '../../types/Logging';
import { sql } from '../util';
import type { WritableDB } from '../Interface';
export const version = 1320;
export function updateToSchemaVersion1320(
currentVersion: number,
db: WritableDB,
logger: LoggerType
): void {
if (currentVersion >= 1320) {
return;
}
db.transaction(() => {
const [query] = sql`
DROP INDEX unprocessed_timestamp;
ALTER TABLE unprocessed
ADD COLUMN receivedAtDate INTEGER DEFAULT 0 NOT NULL;
UPDATE unprocessed
SET receivedAtDate = timestamp;
CREATE INDEX unprocessed_byReceivedAtDate ON unprocessed
(receivedAtDate);
`;
db.exec(query);
db.pragma('user_version = 1320');
})();
logger.info('updateToSchemaVersion1320: success!');
}

View file

@ -107,10 +107,11 @@ import { updateToSchemaVersion1270 } from './1270-normalize-messages';
import { updateToSchemaVersion1280 } from './1280-blob-unprocessed';
import { updateToSchemaVersion1290 } from './1290-int-unprocessed-source-device';
import { updateToSchemaVersion1300 } from './1300-sticker-pack-refs';
import { updateToSchemaVersion1310 } from './1310-muted-fixup';
import {
updateToSchemaVersion1310,
updateToSchemaVersion1320,
version as MAX_VERSION,
} from './1310-muted-fixup';
} from './1320-unprocessed-received-at-date';
import { DataWriter } from '../Server';
function updateToSchemaVersion1(
@ -2089,6 +2090,7 @@ export const SCHEMA_VERSIONS = [
updateToSchemaVersion1300,
updateToSchemaVersion1310,
updateToSchemaVersion1320,
];
export class DBVersionFromFutureError extends Error {

View file

@ -54,6 +54,8 @@ describe('SignalProtocolStore', () => {
let identityKey: KeyPairType;
let testKey: KeyPairType;
const NOW = Date.now();
const unprocessedDefaults = {
type: 1,
messageAgeSec: 1,
@ -73,6 +75,7 @@ describe('SignalProtocolStore', () => {
isEncrypted: true,
content: Buffer.from('content'),
timestamp: NOW,
};
function getSessionRecord(isOpen?: boolean): SessionRecord {
@ -1254,7 +1257,7 @@ describe('SignalProtocolStore', () => {
id: '2-two',
content: Buffer.from('second'),
timestamp: Date.now() + 2,
receivedAtDate: Date.now() + 2,
},
{ zone }
);
@ -1312,7 +1315,7 @@ describe('SignalProtocolStore', () => {
id: '2-two',
content: Buffer.from('second'),
timestamp: 2,
receivedAtDate: 2,
},
{ zone }
);
@ -1436,8 +1439,6 @@ describe('SignalProtocolStore', () => {
});
describe('Not yet processed messages', () => {
const NOW = Date.now();
beforeEach(async () => {
await store.removeAllUnprocessed();
const items = await store.getUnprocessedByIdsAndIncrementAttempts(
@ -1454,7 +1455,7 @@ describe('SignalProtocolStore', () => {
content: Buffer.from('old envelope'),
receivedAtCounter: -1,
timestamp: NOW - 2 * durations.MONTH,
receivedAtDate: NOW - 2 * durations.MONTH,
}),
store.addUnprocessed({
...unprocessedDefaults,
@ -1462,7 +1463,7 @@ describe('SignalProtocolStore', () => {
content: Buffer.from('second'),
receivedAtCounter: 1,
timestamp: NOW + 2,
receivedAtDate: NOW + 2,
}),
store.addUnprocessed({
...unprocessedDefaults,
@ -1470,7 +1471,7 @@ describe('SignalProtocolStore', () => {
content: Buffer.from('third'),
receivedAtCounter: 2,
timestamp: NOW + 3,
receivedAtDate: NOW + 3,
}),
store.addUnprocessed({
...unprocessedDefaults,
@ -1478,7 +1479,7 @@ describe('SignalProtocolStore', () => {
content: Buffer.from('first'),
receivedAtCounter: 0,
timestamp: NOW + 1,
receivedAtDate: NOW + 1,
}),
]);
@ -1501,7 +1502,7 @@ describe('SignalProtocolStore', () => {
id,
timestamp: NOW + 1,
receivedAtDate: NOW + 1,
});
await store.removeUnprocessed(id);
@ -1518,7 +1519,7 @@ describe('SignalProtocolStore', () => {
id: '1-one',
attempts: 10,
timestamp: NOW + 1,
receivedAtDate: NOW + 1,
});
const items = await store.getUnprocessedByIdsAndIncrementAttempts(

View file

@ -65,6 +65,7 @@ describe('unprocessed', function (this: Mocha.Suite) {
sends.push(
alice.sendText(desktop, `hello: ${i}`, {
timestamp: bootstrap.getTimestamp(),
sealed: i % 2 === 0,
})
);
}
@ -91,6 +92,7 @@ describe('unprocessed', function (this: Mocha.Suite) {
.locator(`[data-testid="${alice.device.aci}"] >> "${alice.profileName}"`)
.click();
await page.locator('.module-message__text >> "hello: 4"').waitFor();
await page.locator('.module-message__text >> "hello: 5"').waitFor();
});
});

View file

@ -874,9 +874,8 @@ export default class MessageReceiver
const envelope: ProcessedEnvelope = {
id: item.id,
receivedAtCounter: item.receivedAtCounter ?? item.timestamp,
receivedAtDate:
item.receivedAtCounter == null ? Date.now() : item.timestamp,
receivedAtCounter: item.receivedAtCounter,
receivedAtDate: item.receivedAtDate,
messageAgeSec: item.messageAgeSec,
// Proto.Envelope fields
@ -1189,11 +1188,8 @@ export default class MessageReceiver
sourceServiceId: envelope.sourceServiceId,
sourceDevice: envelope.sourceDevice,
destinationServiceId: envelope.destinationServiceId,
// This field is only used for aging items out of the cache. The original
// envelope's timestamp will be used when retrying this item.
timestamp: envelope.receivedAtDate,
timestamp: envelope.timestamp,
receivedAtDate: envelope.receivedAtDate,
attempts: 0,
isEncrypted: true,
content: envelope.content,