mirrors/signal-desktop
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages 2 Wiki Activity Actions

Storage Service Write: Improved conflict handling

Browse source
This commit is contained in:
Josh Perez 2020-09-29 19:29:11 -04:00 • committed by Josh Perez
parent 866217a724
commit 27759233e4
10 changed files with 297 additions and 134 deletions
Download patch file Download diff file Expand all files Collapse all files

320
ts/services/storage.ts
View file

@ -8,7 +8,6 @@ import {
base64ToArrayBuffer,
deriveStorageItemKey,
deriveStorageManifestKey,
fromEncodedBinaryToArrayBuffer,
} from '../Crypto';
import {
ManifestRecordClass,
@ -29,6 +28,7 @@ import {
toGroupV2Record,
} from './storageRecordOps';
import { ConversationModel } from '../models/conversations';
import { storageJobQueue } from '../util/JobQueue';
const {
eraseStorageServiceStateFromConversations,
@ -37,6 +37,15 @@ const {
let consecutiveStops = 0;
let consecutiveConflicts = 0;
const forcedPushBucket: Array<number> = [];
const validRecordTypes = new Set([
0, // UNKNOWN
1, // CONTACT
2, // GROUPV1
3, // GROUPV2
4, // ACCOUNT
]);
type BackoffType = {
[key: number]: number | undefined;
@ -98,15 +107,6 @@ function generateStorageID(): ArrayBuffer {
return Crypto.getRandomBytes(16);
}
function isGroupV1(conversation: ConversationModel): boolean {
const groupID = conversation.get('groupId');
if (!groupID) {
return false;
}
return fromEncodedBinaryToArrayBuffer(groupID).byteLength === 16;
}
type GeneratedManifestType = {
conversationsToUpdate: Array<{
conversation: ConversationModel;
@ -153,7 +153,7 @@ async function generateManifest(
// eslint-disable-next-line no-await-in-loop
storageRecord.groupV2 = await toGroupV2Record(conversation);
identifier.type = ITEM_TYPE.GROUPV2;
} else if (isGroupV1(conversation)) {
} else if (conversation.isGroupV1()) {
storageRecord = new window.textsecure.protobuf.StorageRecord();
// eslint-disable-next-line no-await-in-loop
storageRecord.groupV1 = await toGroupV1Record(conversation);
@ -166,12 +166,16 @@ async function generateManifest(
}
if (storageRecord) {
const currentStorageID = conversation.get('storageID');
const isNewItem =
isNewManifest || Boolean(conversation.get('needsStorageServiceSync'));
isNewManifest ||
Boolean(conversation.get('needsStorageServiceSync')) ||
!currentStorageID;
const storageID = isNewItem
? arrayBufferToBase64(generateStorageID())
: conversation.get('storageID');
: currentStorageID;
let storageItem;
try {
@ -217,6 +221,27 @@ async function generateManifest(
// When updating the manifest, ensure all "unknown" keys are added to the
// new manifest, so we don't inadvertently delete something we don't understand
unknownRecordsArray.forEach((record: UnknownRecord) => {
if (validRecordTypes.has(record.itemType)) {
return;
}
const identifier = new window.textsecure.protobuf.ManifestRecord.Identifier();
identifier.type = record.itemType;
identifier.raw = base64ToArrayBuffer(record.storageID);
manifestRecordKeys.add(identifier);
});
const recordsWithErrors =
window.storage.get('storage-service-error-records') || [];
window.log.info(
`storageService.generateManifest: adding ${recordsWithErrors.length} records that had errors in the previous merge`
);
// These records failed to merge in the previous fetchManifest, but we still
// need to include them so that the manifest is complete
recordsWithErrors.forEach((record: UnknownRecord) => {
const identifier = new window.textsecure.protobuf.ManifestRecord.Identifier();
identifier.type = record.itemType;
identifier.raw = base64ToArrayBuffer(record.storageID);
@ -334,6 +359,11 @@ async function uploadManifest(
throw new Error('storageService.uploadManifest: We are offline!');
}
if (newItems.size === 0 && deleteKeys.length === 0) {
window.log.info('storageService.uploadManifest: nothing to upload');
return;
}
const credentials = window.storage.get('storageCredentials');
try {
window.log.info(
@ -345,7 +375,7 @@ async function uploadManifest(
writeOperation.insertItem = Array.from(newItems);
writeOperation.deleteKey = deleteKeys;
window.log.info('storageService.uploadManifest: uploading...');
window.log.info('storageService.uploadManifest: uploading...', version);
await window.textsecure.messaging.modifyStorageRecords(
writeOperation.toArrayBuffer(),
{
@ -382,7 +412,7 @@ async function uploadManifest(
consecutiveConflicts += 1;
window.log.info(
`storageService.uploadManifest: Conflict found, running sync job times(${consecutiveConflicts})`
`storageService.uploadManifest: Conflict found with v${version}, running sync job times(${consecutiveConflicts})`
);
throw err;
@ -527,6 +557,7 @@ type MergeableItemType = {
type MergedRecordType = UnknownRecord & {
hasConflict: boolean;
hasError: boolean;
isUnsupported: boolean;
};
@ -539,6 +570,7 @@ async function mergeRecord(
let hasConflict = false;
let isUnsupported = false;
let hasError = false;
try {
if (itemType === ITEM_TYPE.UNKNOWN) {
@ -565,6 +597,7 @@ async function mergeRecord(
);
}
} catch (err) {
hasError = true;
window.log.error(
'storageService.mergeRecord: merging record failed',
storageID,
@ -574,6 +607,7 @@ async function mergeRecord(
return {
hasConflict,
hasError,
isUnsupported,
itemType,
storageID,
@ -607,7 +641,10 @@ async function processManifest(
window.storage.get('storage-service-unknown-records') || [];
unknownRecordsArray.forEach((record: UnknownRecord) => {
localKeys.push(record.storageID);
// Do not include any unknown records that we already support
if (!validRecordTypes.has(record.itemType)) {
localKeys.push(record.storageID);
}
});
window.log.info(
@ -616,9 +653,14 @@ async function processManifest(
const remoteKeys = Array.from(remoteKeysTypeMap.keys());
const remoteOnly = remoteKeys.filter(
(key: string) => !localKeys.includes(key)
);
const remoteOnlySet: Set<string> = new Set();
remoteKeys.forEach((key: string) => {
if (!localKeys.includes(key)) {
remoteOnlySet.add(key);
}
});
const remoteOnly = Array.from(remoteOnlySet);
window.log.info(
`storageService.processManifest: remoteOnly.length ${remoteOnly.length}`
@ -694,7 +736,7 @@ async function processManifest(
storageRecord,
};
},
{ concurrency: 50 }
{ concurrency: 5 }
);
// Merge Account records last
@ -713,7 +755,7 @@ async function processManifest(
concurrency: 5,
});
window.log.info(
`storageService.processManifest: Merged ${mergedRecords.length} records`
`storageService.processManifest: Processed ${mergedRecords.length} records`
);
const unknownRecords: Map<string, UnknownRecord> = new Map();
@ -721,21 +763,89 @@ async function processManifest(
unknownRecords.set(record.storageID, record);
});
const hasConflict = mergedRecords.some((mergedRecord: MergedRecordType) => {
const recordsWithErrors: Array<UnknownRecord> = [];
let hasConflict = false;
mergedRecords.forEach((mergedRecord: MergedRecordType) => {
if (mergedRecord.isUnsupported) {
unknownRecords.set(mergedRecord.storageID, {
itemType: mergedRecord.itemType,
storageID: mergedRecord.storageID,
});
} else if (mergedRecord.hasError) {
recordsWithErrors.push({
itemType: mergedRecord.itemType,
storageID: mergedRecord.storageID,
});
}
return mergedRecord.hasConflict;
hasConflict = hasConflict || mergedRecord.hasConflict;
});
window.storage.put(
'storage-service-unknown-records',
Array.from(unknownRecords.values())
// Filter out all the unknown records we're already supporting
const newUnknownRecords = Array.from(unknownRecords.values()).filter(
(record: UnknownRecord) => !validRecordTypes.has(record.itemType)
);
window.log.info(
`storageService.processManifest: Found ${newUnknownRecords.length} unknown records`
);
window.storage.put('storage-service-unknown-records', newUnknownRecords);
window.log.info(
`storageService.processManifest: Found ${recordsWithErrors.length} records with errors`
);
window.storage.put('storage-service-error-records', recordsWithErrors);
const now = Date.now();
// if the remote only keys are larger or equal to our local keys then it
// was likely a forced push of storage service. We keep track of these
// merges so that we can detect possible infinite loops
if (remoteOnly.length >= localKeys.length) {
window.log.info(
'storageService.processManifest: remote manifest was likely force pushed',
now
);
forcedPushBucket.push(now);
// we need to check our conversations because maybe all of them were not
// updated properly, for those that weren't we'll clear their storage
// key so that they can be included in the next update
window.getConversations().forEach((conversation: ConversationModel) => {
const storageID = conversation.get('storageID');
if (storageID && !remoteOnlySet.has(storageID)) {
window.log.info(
'storageService.processManifest: conversation was not included in remote force push, clearing storageID',
conversation.debugID(),
storageID
);
conversation.unset('storageID');
}
});
if (forcedPushBucket.length >= 3) {
const [firstMostRecentForcedPush] = forcedPushBucket;
if (now - firstMostRecentForcedPush < 5 * MINUTE) {
window.log.info(
'storageService.processManifest: thrasing? Backing off'
);
const error = new Error();
error.code = 'E_BACKOFF';
throw error;
}
window.log.info(
'storageService.processManifest: thrash timestamp of first -> now',
firstMostRecentForcedPush,
now
);
forcedPushBucket.shift();
}
}
if (hasConflict) {
window.log.info(
'storageService.processManifest: Conflict found, uploading changes'
@ -756,24 +866,20 @@ async function processManifest(
return false;
}
// Exported functions
export async function runStorageServiceSyncJob(): Promise<void> {
async function sync(): Promise<void> {
if (!isEnabled('desktop.storage')) {
window.log.info(
'storageService.runStorageServiceSyncJob: Not starting desktop.storage is falsey'
'storageService.sync: Not starting desktop.storage is falsey'
);
return;
}
if (!window.storage.get('storageKey')) {
throw new Error(
'storageService.runStorageServiceSyncJob: Cannot start; no storage key!'
);
throw new Error('storageService.sync: Cannot start; no storage key!');
}
window.log.info('storageService.runStorageServiceSyncJob: starting...');
window.log.info('storageService.sync: starting...');
try {
const localManifestVersion = window.storage.get('manifestVersion') || 0;
@ -781,45 +887,103 @@ export async function runStorageServiceSyncJob(): Promise<void> {
// Guarding against no manifests being returned, everything should be ok
if (!manifest) {
window.log.info(
'storageService.runStorageServiceSyncJob: no new manifest'
);
window.log.info('storageService.sync: no new manifest');
return;
}
const version = manifest.version.toNumber();
window.log.info(
`storageService.runStorageServiceSyncJob: manifest versions - previous: ${localManifestVersion}, current: ${version}`
`storageService.sync: manifest versions - previous: ${localManifestVersion}, current: ${version}`
);
window.storage.put('manifestVersion', version);
const hasConflicts = await processManifest(manifest);
if (hasConflicts) {
await storageServiceUploadJob();
await upload();
}
window.storage.put('manifestVersion', version);
} catch (err) {
window.log.error(
`storageService.runStorageServiceSyncJob: error processing manifest ${
`storageService.sync: error processing manifest ${
err && err.stack ? err.stack : String(err)
}`
);
// When we're told to backoff, backoff to the max which should be
// ~5 minutes. If this job was running inside a queue it'll probably time
// out.
if (err.code === 'E_BACKOFF') {
await backOff(9001);
}
}
window.log.info('storageService.runStorageServiceSyncJob: complete');
window.log.info('storageService.sync: complete');
}
// Note: this function must be called at startup once we handle unknown records
// of a certain type. This way once the runStorageServiceSyncJob function runs
// it'll pick up the new storage IDs and process them accordingly.
export function handleUnknownRecords(itemType: number): void {
const unknownRecordsArray =
window.storage.get('storage-service-unknown-records') || [];
const newUnknownRecords = unknownRecordsArray.filter(
(record: UnknownRecord) => record.itemType !== itemType
async function upload(): Promise<void> {
if (!isEnabled('desktop.storage')) {
window.log.info(
'storageService.upload: Not starting desktop.storage is falsey'
);
return;
}
if (!isEnabled('desktop.storageWrite')) {
window.log.info(
'storageService.upload: Not starting desktop.storageWrite is falsey'
);
return;
}
if (!window.textsecure.messaging) {
throw new Error('storageService.upload: We are offline!');
}
if (!window.storage.get('storageKey')) {
// requesting new keys runs the sync job which will detect the conflict
// and re-run the upload job once we're merged and up-to-date.
window.log.info(
'storageService.upload: no storageKey, requesting new keys'
);
consecutiveStops = 0;
await window.textsecure.messaging.sendRequestKeySyncMessage();
return;
}
const localManifestVersion = window.storage.get('manifestVersion') || 0;
const version = Number(localManifestVersion) + 1;
window.log.info(
'storageService.upload: will update to manifest version',
version
);
window.storage.put('storage-service-unknown-records', newUnknownRecords);
try {
const generatedManifest = await generateManifest(version);
await uploadManifest(version, generatedManifest);
} catch (err) {
if (err.code === 409) {
await backOff(consecutiveConflicts);
window.log.info('storageService.upload: pushing sync on the queue');
// The sync job will check for conflicts and as part of that conflict
// check if an item needs sync and doesn't match with the remote record
// it'll kick off another upload.
setTimeout(runStorageServiceSyncJob);
return;
}
window.log.error(
'storageService.upload',
err && err.stack ? err.stack : String(err)
);
}
}
let storageServiceEnabled = false;
export function enableStorageService(): void {
storageServiceEnabled = true;
}
// Note: this function is meant to be called before ConversationController is hydrated.
@ -835,56 +999,24 @@ export async function eraseAllStorageServiceState(): Promise<void> {
window.log.info('storageService.eraseAllStorageServiceState: complete');
}
async function nondebouncedStorageServiceUploadJob(): Promise<void> {
if (!isEnabled('desktop.storage')) {
export const storageServiceUploadJob = _.debounce(() => {
if (!storageServiceEnabled) {
window.log.info(
'storageService.storageServiceUploadJob: Not starting desktop.storage is falsey'
'storageService.storageServiceUploadJob: called before enabled'
);
return;
}
if (!isEnabled('desktop.storageWrite')) {
window.log.info(
'storageService.storageServiceUploadJob: Not starting desktop.storageWrite is falsey'
);
return;
}
if (!window.textsecure.messaging) {
throw new Error('storageService.storageServiceUploadJob: We are offline!');
}
storageJobQueue(upload, `upload v${window.storage.get('manifestVersion')}`);
}, 500);
if (!window.storage.get('storageKey')) {
// requesting new keys runs the sync job which will detect the conflict
// and re-run the upload job once we're merged and up-to-date.
export const runStorageServiceSyncJob = _.debounce(() => {
if (!storageServiceEnabled) {
window.log.info(
'storageService.storageServiceUploadJob: no storageKey, requesting new keys'
'storageService.runStorageServiceSyncJob: called before enabled'
);
consecutiveStops = 0;
await window.textsecure.messaging.sendRequestKeySyncMessage();
return;
}
const localManifestVersion = window.storage.get('manifestVersion') || 0;
const version = Number(localManifestVersion) + 1;
window.log.info(
'storageService.storageServiceUploadJob: will update to manifest version',
version
);
try {
await uploadManifest(version, await generateManifest(version));
} catch (err) {
if (err.code === 409) {
await backOff(consecutiveConflicts);
await runStorageServiceSyncJob();
}
}
}
export const storageServiceUploadJob = _.debounce(
nondebouncedStorageServiceUploadJob,
500
);
storageJobQueue(sync, `sync v${window.storage.get('manifestVersion')}`);
}, 500);