matrix/matrix-js-sdk
1
0
Fork 0
mirror of https://github.com/matrix-org/matrix-js-sdk.git synced 2025-11-26 17:03:12 +03:00
Code Activity

Initial attempt at device tracking -> indexeddb

Browse Source 
* Message sending works again, but

 * Marking multiple devices known (ie. 'send anyway') is very slow
   because it writes all device info out each time
 * Support for non-indexedb stores not written yet
 * No migration
This commit is contained in:
David Baker
2018-01-11 18:10:19 +00:00
parent eeb97f5b66
commit 83e4aa2755
5 changed files with 253 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

193
src/crypto/DeviceList.js
View File

@@ -25,6 +25,7 @@ import Promise from 'bluebird';
import DeviceInfo from './deviceinfo';
import olmlib from './olmlib';
import IndexedDBCryptoStore from './store/indexeddb-crypto-store';
/* State transition diagram for DeviceList._deviceTrackingStatus
@@ -58,26 +59,95 @@ const TRACKING_STATUS_UP_TO_DATE = 3;
* @alias module:crypto/DeviceList
*/
export default class DeviceList {
constructor(baseApis, sessionStore, olmDevice) {
this._sessionStore = sessionStore;
this._serialiser = new DeviceListUpdateSerialiser(
baseApis, sessionStore, olmDevice,
);
constructor(baseApis, cryptoStore, olmDevice) {
this._cryptoStore = cryptoStore;
// userId -> {
// deviceId -> {
// [device info]
// }
// }
this._devices = null;
// which users we are tracking device status for.
// userId -> TRACKING_STATUS_*
this._deviceTrackingStatus = sessionStore.getEndToEndDeviceTrackingStatus() || {};
this._deviceTrackingStatus = null; // loaded from storage in load()
// The 'next_batch' sync token at the point the data was writen,
// ie. a token represtenting the point immediately after the
// moment represented by the snapshot in the db.
this._syncToken = null;
this._serialiser = new DeviceListUpdateSerialiser(
baseApis, olmDevice,
);
// userId -> promise
this._keyDownloadsInProgressByUser = {};
// Set whenever changes are made other than setting the sync token
this._dirty = false;
}
/**
* Load the device tracking state from storage
*/
async load() {
await this._cryptoStore.doTxn(
'readonly', [IndexedDBCryptoStore.STORE_DEVICE_DATA], (txn) => {
this._cryptoStore.getEndToEndDeviceData(txn, (deviceData) => {
this._devices = deviceData ? deviceData.devices : {},
this._deviceTrackingStatus = deviceData ? deviceData.trackingStatus : {};
this._syncToken = deviceData ? deviceData.syncToken : null;
});
},
);
for (const u of Object.keys(this._deviceTrackingStatus)) {
// if a download was in progress when we got shut down, it isn't any more.
if (this._deviceTrackingStatus[u] == TRACKING_STATUS_DOWNLOAD_IN_PROGRESS) {
this._deviceTrackingStatus[u] = TRACKING_STATUS_PENDING_DOWNLOAD;
}
}
}
// userId -> promise
this._keyDownloadsInProgressByUser = {};
/**
* Save the device tracking state to storage, if any changes are
* pending other than updating the sync token
* Before calling this, the caller must ensure that the state it
* has set this object to is consistent, ie. the appropriate sync
* token has been set with setSyncToken for any device updates that
* have occurred.
*/
async saveIfDirty() {
if (!this._dirty) return;
await this._cryptoStore.doTxn(
'readwrite', [IndexedDBCryptoStore.STORE_DEVICE_DATA], (txn) => {
this._cryptoStore.storeEndToEndDeviceData({
devices: this._devices,
trackingStatus: this._deviceTrackingStatus,
syncToken: this._syncToken,
}, txn);
},
);
this._dirty = false;
}
this.lastKnownSyncToken = null;
/**
* Gets the current sync token
*
* @return {string} The sync token
*/
getSyncToken() {
return this._syncToken;
}
/**
* Sets the current sync token
*
* @param {string} st The sync token
*/
setSyncToken(st) {
this._syncToken = st;
}
/**
@@ -152,7 +222,7 @@ export default class DeviceList {
* managed to get a list of devices for this user yet.
*/
getStoredDevicesForUser(userId) {
const devs = this._sessionStore.getEndToEndDevicesForUser(userId);
const devs = this._devices[userId];
if (!devs) {
return null;
}
@@ -165,6 +235,22 @@ export default class DeviceList {
return res;
}
/**
* Get the stored device data for a user, in raw object form
*
* @param {string} userId the user to get data for
*
* @return {Object} userId->deviceId->{object} devices, or null if
* there is no data for this user.
*/
getRawStoredDevicesForUser(userId) {
const devs = this._devices[userId];
if (!devs) {
return null;
}
return devs;
}
/**
* Get the stored keys for a single device
*
@@ -175,7 +261,7 @@ export default class DeviceList {
* if we don't know about this device
*/
getStoredDevice(userId, deviceId) {
const devs = this._sessionStore.getEndToEndDevicesForUser(userId);
const devs = this._devices[userId];
if (!devs || !devs[deviceId]) {
return undefined;
}
@@ -200,7 +286,7 @@ export default class DeviceList {
return null;
}
const devices = this._sessionStore.getEndToEndDevicesForUser(userId);
const devices = this._devices[userId];
if (!devices) {
return null;
}
@@ -229,6 +315,14 @@ export default class DeviceList {
return null;
}
/**
* Replaces the list of devices for a user with the given device list
*/
storeDevicesForUser(u, devs) {
this._devices[u] = devs;
this._dirty = true;
}
/**
* flag the given user for device-list tracking, if they are not already.
*
@@ -254,8 +348,8 @@ export default class DeviceList {
this._deviceTrackingStatus[userId] = TRACKING_STATUS_PENDING_DOWNLOAD;
}
// we don't yet persist the tracking status, since there may be a lot
// of calls; instead we wait for the forthcoming
// refreshOutdatedDeviceLists.
// of calls; we save all data together once the sync is done
this._dirty = true;
}
/**
@@ -273,8 +367,9 @@ export default class DeviceList {
this._deviceTrackingStatus[userId] = TRACKING_STATUS_NOT_TRACKED;
}
// we don't yet persist the tracking status, since there may be a lot
// of calls; instead we wait for the forthcoming
// refreshOutdatedDeviceLists.
// of calls; we save all data together once the sync is done
this._dirty = true;
}
@@ -295,8 +390,9 @@ export default class DeviceList {
this._deviceTrackingStatus[userId] = TRACKING_STATUS_PENDING_DOWNLOAD;
}
// we don't yet persist the tracking status, since there may be a lot
// of calls; instead we wait for the forthcoming
// refreshOutdatedDeviceLists.
// of calls; we save all data together once the sync is done
this._dirty = true;
}
/**
@@ -318,6 +414,8 @@ export default class DeviceList {
* is no need to wait for this (it's mostly for the unit tests).
*/
refreshOutdatedDeviceLists() {
this.saveIfDirty();
const usersToDownload = [];
for (const userId of Object.keys(this._deviceTrackingStatus)) {
const stat = this._deviceTrackingStatus[userId];
@@ -326,10 +424,6 @@ export default class DeviceList {
}
}
// we didn't persist the tracking status during
// invalidateUserDeviceList, so do it now.
this._persistDeviceTrackingStatus();
return this._doKeyDownload(usersToDownload);
}
@@ -352,14 +446,14 @@ export default class DeviceList {
}
const prom = this._serialiser.updateDevicesForUsers(
users, this.lastKnownSyncToken,
).then(() => {
finished(true);
this._devices, users, this._syncToken,
).then((newDevices) => {
finished(newDevices);
}, (e) => {
console.error(
'Error downloading keys for ' + users + ":", e,
);
finished(false);
finished(null);
throw e;
});
@@ -371,7 +465,7 @@ export default class DeviceList {
}
});
const finished = (success) => {
const finished = (newDevices) => {
users.forEach((u) => {
// we may have queued up another download request for this user
// since we started this request. If that happens, we should
@@ -384,25 +478,23 @@ export default class DeviceList {
delete this._keyDownloadsInProgressByUser[u];
const stat = this._deviceTrackingStatus[u];
if (stat == TRACKING_STATUS_DOWNLOAD_IN_PROGRESS) {
if (success) {
if (newDevices) {
// we didn't get any new invalidations since this download started:
// this user's device list is now up to date.
this._deviceTrackingStatus[u] = TRACKING_STATUS_UP_TO_DATE;
this._devices[u] = newDevices[u];
this._dirty = true;
console.log("Device list for", u, "now up to date");
} else {
this._deviceTrackingStatus[u] = TRACKING_STATUS_PENDING_DOWNLOAD;
}
}
});
this._persistDeviceTrackingStatus();
this.saveIfDirty();
};
return prom;
}
_persistDeviceTrackingStatus() {
this._sessionStore.storeEndToEndDeviceTrackingStatus(this._deviceTrackingStatus);
}
}
/**
@@ -415,9 +507,8 @@ export default class DeviceList {
* time (and queuing other requests up).
*/
class DeviceListUpdateSerialiser {
constructor(baseApis, sessionStore, olmDevice) {
constructor(baseApis, olmDevice) {
this._baseApis = baseApis;
this._sessionStore = sessionStore;
this._olmDevice = olmDevice;
this._downloadInProgress = false;
@@ -431,14 +522,15 @@ class DeviceListUpdateSerialiser {
// non-null indicates that we have users queued for download.
this._queuedQueryDeferred = null;
// sync token to be used for the next query: essentially the
// most recent one we know about
this._nextSyncToken = null;
this._devices = null; // the complete device list
this._updatedDevices = null; // device list updates we've fetched
}
/**
* Make a key query request for the given users
*
* @param {object} devices The current device list
*
* @param {String[]} users list of user ids
*
* @param {String} syncToken sync token to pass in the query request, to
@@ -446,13 +538,12 @@ class DeviceListUpdateSerialiser {
*
* @return {module:client.Promise} resolves when all the users listed have
* been updated. rejects if there was a problem updating any of the
* users.
* users. Returns a fresh device list object for the users queried.
*/
updateDevicesForUsers(users, syncToken) {
updateDevicesForUsers(devices, users, syncToken) {
users.forEach((u) => {
this._keyDownloadsQueuedByUser[u] = true;
});
this._nextSyncToken = syncToken;
if (!this._queuedQueryDeferred) {
this._queuedQueryDeferred = Promise.defer();
@@ -464,11 +555,13 @@ class DeviceListUpdateSerialiser {
return this._queuedQueryDeferred.promise;
}
this._devices = devices;
this._updatedDevices = {};
// start a new download.
return this._doQueuedQueries();
return this._doQueuedQueries(syncToken);
}
_doQueuedQueries() {
_doQueuedQueries(syncToken) {
if (this._downloadInProgress) {
throw new Error(
"DeviceListUpdateSerialiser._doQueuedQueries called with request active",
@@ -484,8 +577,8 @@ class DeviceListUpdateSerialiser {
this._downloadInProgress = true;
const opts = {};
if (this._nextSyncToken) {
opts.token = this._nextSyncToken;
if (syncToken) {
opts.token = syncToken;
}
this._baseApis.downloadKeysForUsers(
@@ -510,11 +603,11 @@ class DeviceListUpdateSerialiser {
console.log('Completed key download for ' + downloadUsers);
this._downloadInProgress = false;
deferred.resolve();
deferred.resolve(this._updatedDevices);
// if we have queued users, fire off another request.
if (this._queuedQueryDeferred) {
this._doQueuedQueries();
this._doQueuedQueries(syncToken);
}
}, (e) => {
console.warn('Error downloading keys for ' + downloadUsers + ':', e);
@@ -530,7 +623,7 @@ class DeviceListUpdateSerialiser {
// map from deviceid -> deviceinfo for this user
const userStore = {};
const devs = this._sessionStore.getEndToEndDevicesForUser(userId);
const devs = this._devices[userId];
if (devs) {
Object.keys(devs).forEach((deviceId) => {
const d = DeviceInfo.fromStorage(devs[deviceId], deviceId);
@@ -548,9 +641,7 @@ class DeviceListUpdateSerialiser {
storage[deviceId] = userStore[deviceId].toStorage();
});
this._sessionStore.storeEndToEndDevicesForUser(
userId, storage,
);
this._updatedDevices[userId] = storage;
}
}

115
src/crypto/index.js
View File

@@ -69,7 +69,7 @@ function Crypto(baseApis, sessionStore, userId, deviceId,
this._cryptoStore = cryptoStore;
this._olmDevice = new OlmDevice(sessionStore, cryptoStore);
this._deviceList = new DeviceList(baseApis, sessionStore, this._olmDevice);
this._deviceList = new DeviceList(baseApis, cryptoStore, this._olmDevice);
// the last time we did a check for the number of one-time-keys on the
// server.
@@ -128,6 +128,7 @@ Crypto.prototype.init = async function() {
}
await this._olmDevice.init();
await this._deviceList.load();
// build our device keys: these will later be uploaded
this._deviceKeys["ed25519:" + this._deviceId] =
@@ -135,7 +136,7 @@ Crypto.prototype.init = async function() {
this._deviceKeys["curve25519:" + this._deviceId] =
this._olmDevice.deviceCurve25519Key;
let myDevices = this._sessionStore.getEndToEndDevicesForUser(
let myDevices = this._deviceList.getRawStoredDevicesForUser(
this._userId,
);
@@ -153,9 +154,10 @@ Crypto.prototype.init = async function() {
};
myDevices[this._deviceId] = deviceInfo;
this._sessionStore.storeEndToEndDevicesForUser(
this._deviceList.storeDevicesForUser(
this._userId, myDevices,
);
this._deviceList.saveIfDirty();
}
};
@@ -456,7 +458,7 @@ Crypto.prototype.getStoredDevice = function(userId, deviceId) {
Crypto.prototype.setDeviceVerification = async function(
userId, deviceId, verified, blocked, known,
) {
const devices = this._sessionStore.getEndToEndDevicesForUser(userId);
const devices = this._deviceList.getRawStoredDevicesForUser(userId);
if (!devices || !devices[deviceId]) {
throw new Error("Unknown device " + userId + ":" + deviceId);
}
@@ -484,7 +486,8 @@ Crypto.prototype.setDeviceVerification = async function(
if (dev.verified !== verificationStatus || dev.known !== knownStatus) {
dev.verified = verificationStatus;
dev.known = knownStatus;
this._sessionStore.storeEndToEndDevicesForUser(userId, devices);
this._deviceList.storeDevicesForUser(userId, devices);
this._deviceList.saveIfDirty();
}
return DeviceInfo.fromStorage(dev, deviceId);
};
@@ -812,21 +815,31 @@ Crypto.prototype.decryptEvent = function(event) {
* @param {Object} deviceLists device_lists field from /sync, or response from
* /keys/changes
*/
Crypto.prototype.handleDeviceListChanges = async function(deviceLists) {
if (deviceLists.changed && Array.isArray(deviceLists.changed)) {
deviceLists.changed.forEach((u) => {
this._deviceList.invalidateUserDeviceList(u);
});
}
Crypto.prototype.handleDeviceListChanges = async function(syncData, syncDeviceLists) {
// No point processing device list changes for initial syncs: they'd be meaningless
// since the server doesn't know what point we were were at previously. We'll either
// get the complete list of changes for the interval or invalidate everything in
// onSyncComplete
if (!syncData.oldSyncToken) return;
if (deviceLists.left && Array.isArray(deviceLists.left)) {
deviceLists.left.forEach((u) => {
this._deviceList.stopTrackingDeviceList(u);
});
if (syncData.oldSyncToken === this._deviceList.getSyncToken()) {
// the point the db is at matches where the sync started from, so
// we can safely write the changes
this._evalDeviceListChanges(syncDeviceLists);
} else {
// the db is at a different point to where this sync started from, so
// additionally fetch the changes between where the db is and where the
// sync started
console.log(
"Device list sync gap detected - fetching key changes between " +
this._deviceList.getSyncToken() + " and " + syncData.oldSyncToken,
);
const gapDeviceLists = await this._baseApis.getKeyChanges(
this._deviceList.getSyncToken(), syncData.oldSyncToken,
);
this._evalDeviceListChanges(gapDeviceLists);
this._evalDeviceListChanges(syncDeviceLists);
}
// don't flush the outdated device list yet - we do it once we finish
// processing the sync.
};
/**
@@ -890,36 +903,15 @@ Crypto.prototype.onSyncCompleted = async function(syncData) {
const nextSyncToken = syncData.nextSyncToken;
if (!syncData.oldSyncToken) {
console.log("Completed initial sync");
// if we have a deviceSyncToken, we can tell the deviceList to
// invalidate devices which have changed since then.
const oldSyncToken = this._sessionStore.getEndToEndDeviceSyncToken();
if (oldSyncToken !== null) {
try {
await this._invalidateDeviceListsSince(
oldSyncToken, nextSyncToken,
);
} catch (e) {
// if that failed, we fall back to invalidating everyone.
console.warn("Error fetching changed device list", e);
// If we have a stored device sync token, we could request the complete
// list of device changes from the server here to get our device list up
// to date. This case should be relatively rare though (only when you hit
// 'clear cache and reload' in practice) so we just invalidate everything.
console.log("invalidating all device list caches after inital sync");
this._deviceList.invalidateAllDeviceLists();
}
} else {
// otherwise, we have to invalidate all devices for all users we
// are tracking.
console.log("Completed first initialsync; invalidating all " +
"device list caches");
this._deviceList.invalidateAllDeviceLists();
}
}
// we can now store our sync token so that we can get an update on
// restart rather than having to invalidate everyone.
//
// (we don't really need to do this on every sync - we could just
// do it periodically)
this._sessionStore.storeEndToEndDeviceSyncToken(nextSyncToken);
this._deviceList.setSyncToken(syncData.nextSyncToken);
this._deviceList.saveIfDirty();
// catch up on any new devices we got told about during the sync.
this._deviceList.lastKnownSyncToken = nextSyncToken;
@@ -936,25 +928,24 @@ Crypto.prototype.onSyncCompleted = async function(syncData) {
};
/**
* Ask the server which users have new devices since a given token,
* and invalidate them
* Trigger the appropriate invalidations and removes for a given
* device list
*
* @param {String} oldSyncToken
* @param {String} lastKnownSyncToken
*
* Returns a Promise which resolves once the query is complete. Rejects if the
* keyChange query fails.
* @param {Object} deviceLists device_lists field from /sync, or response from
* /keys/changes
*/
Crypto.prototype._invalidateDeviceListsSince = async function(
oldSyncToken, lastKnownSyncToken,
) {
const r = await this._baseApis.getKeyChanges(
oldSyncToken, lastKnownSyncToken,
);
Crypto.prototype._evalDeviceListChanges = async function(deviceLists) {
if (deviceLists.changed && Array.isArray(deviceLists.changed)) {
deviceLists.changed.forEach((u) => {
this._deviceList.invalidateUserDeviceList(u);
});
}
console.log("got key changes since", oldSyncToken, ":", r);
await this.handleDeviceListChanges(r);
if (deviceLists.left && Array.isArray(deviceLists.left)) {
deviceLists.left.forEach((u) => {
this._deviceList.stopTrackingDeviceList(u);
});
}
};
/**

22
src/crypto/store/indexeddb-crypto-store-backend.js
View File

@@ -1,7 +1,7 @@
import Promise from 'bluebird';
import utils from '../../utils';
export const VERSION = 4;
export const VERSION = 5;
/**
* Implementation of a CryptoStore which is backed by an existing
@@ -391,6 +391,23 @@ export class Backend {
});
}
getEndToEndDeviceData(txn, func) {
const objectStore = txn.objectStore("device_data");
const getReq = objectStore.get("-");
getReq.onsuccess = function() {
try {
func(getReq.result || null);
} catch (e) {
abortWithException(txn, e);
}
};
}
storeEndToEndDeviceData(deviceData, txn) {
const objectStore = txn.objectStore("device_data");
objectStore.put(deviceData, "-");
}
doTxn(mode, stores, func) {
const txn = this._db.transaction(stores, mode);
const promise = promiseifyTxn(txn);
@@ -423,6 +440,9 @@ export function upgradeDatabase(db, oldVersion) {
keyPath: ["senderCurve25519Key", "sessionId"],
});
}
if (oldVersion < 5) {
db.createObjectStore("device_data");
}
// Expand as needed.
}

25
src/crypto/store/indexeddb-crypto-store.js
View File

@@ -358,6 +358,30 @@ export default class IndexedDBCryptoStore {
);
}
// End-to-end device tracking
/**
* Store the state of all tracked devices
* This contains devices for each user, a tracking state for each user
* and a sync token matching the point in time the snapshot represents.
* These all need to be written out in full each time such that the snapshot
* is always consistent, so they are stored in one object.
*
* @param {Object} deviceData
*/
storeEndToEndDeviceData(deviceData, txn) {
this._backendPromise.value().storeEndToEndDeviceData(deviceData, txn);
}
/**
* Get the state of all tracked devices
*
* @param {*} txn An active transaction. See doTxn().
*/
getEndToEndDeviceData(txn, func) {
return this._backendPromise.value().getEndToEndDeviceData(txn, func);
}
/**
* Perform a transaction on the crypto store. Any store methods
* that require a transaction (txn) object to be passed in may
@@ -389,3 +413,4 @@ export default class IndexedDBCryptoStore {
IndexedDBCryptoStore.STORE_ACCOUNT = 'account';
IndexedDBCryptoStore.STORE_SESSIONS = 'sessions';
IndexedDBCryptoStore.STORE_INBOUND_GROUP_SESSIONS = 'inbound_group_sessions';
IndexedDBCryptoStore.STORE_DEVICE_DATA = 'device_data';

22
src/sync.js
View File

@@ -621,8 +621,14 @@ SyncApi.prototype._sync = async function(syncOptions) {
await client.store.setSyncData(data);
}
const syncEventData = {
oldSyncToken: syncToken,
nextSyncToken: data.next_batch,
catchingUp: this._catchingUp,
};
try {
await this._processSyncResponse(syncToken, data, isCachedResponse);
await this._processSyncResponse(syncEventData, data, isCachedResponse);
} catch(e) {
// log the exception with stack if we have it, else fall back
// to the plain description
@@ -630,12 +636,6 @@ SyncApi.prototype._sync = async function(syncOptions) {
}
// emit synced events
const syncEventData = {
oldSyncToken: syncToken,
nextSyncToken: data.next_batch,
catchingUp: this._catchingUp,
};
if (!syncOptions.hasSyncedBefore) {
this._updateSyncState("PREPARED", syncEventData);
syncOptions.hasSyncedBefore = true;
@@ -708,7 +708,7 @@ SyncApi.prototype._onSyncError = function(err, syncOptions) {
* @param {bool} isCachedResponse True if this response is from our local cache
*/
SyncApi.prototype._processSyncResponse = async function(
syncToken, data, isCachedResponse,
syncEventData, data, isCachedResponse,
) {
const client = this.client;
const self = this;
@@ -950,7 +950,7 @@ SyncApi.prototype._processSyncResponse = async function(
self._deregisterStateListeners(room);
room.resetLiveTimeline(
joinObj.timeline.prev_batch,
self.opts.canResetEntireTimeline(room.roomId) ? null : syncToken,
self.opts.canResetEntireTimeline(room.roomId) ? null : syncEventData.oldSyncToken,
);
// We have to assume any gap in any timeline is
@@ -1034,7 +1034,7 @@ SyncApi.prototype._processSyncResponse = async function(
// in the timeline relative to ones paginated in by /notifications.
// XXX: we could fix this by making EventTimeline support chronological
// ordering... but it doesn't, right now.
if (syncToken && this._notifEvents.length) {
if (syncEventData.oldSyncToken && this._notifEvents.length) {
this._notifEvents.sort(function(a, b) {
return a.getTs() - b.getTs();
});
@@ -1046,7 +1046,7 @@ SyncApi.prototype._processSyncResponse = async function(
// Handle device list updates
if (data.device_lists) {
if (this.opts.crypto) {
await this.opts.crypto.handleDeviceListChanges(data.device_lists);
await this.opts.crypto.handleDeviceListChanges(syncEventData, data.device_lists);
} else {
// FIXME if we *don't* have a crypto module, we still need to
// invalidate the device lists. But that would require a