Page MenuHomePhabricator
Differential D3651

[services] Backup - Send log - finish read request logic
ClosedPublic
Actions

Authored by • karol on Apr 7 2022, 7:22 AM.
Tags
None
Referenced Files
F5058310: D3651.id11399.diff
Wed, Mar 26, 4:24 AM
Unknown Object (File)
Wed, Feb 26, 11:26 AM
Unknown Object (File)
Tue, Feb 25, 1:56 PM
Unknown Object (File)
Feb 21 2025, 8:39 PM
Unknown Object (File)
Feb 16 2025, 6:57 PM
Unknown Object (File)
Feb 8 2025, 6:50 AM
Unknown Object (File)
Feb 5 2025, 5:05 AM
Unknown Object (File)
Feb 3 2025, 5:59 AM
View All Files
Subscribers
• abosh
• adrian
ashoat
atul
• benschac
tomek

Details

Reviewers
tomek
• max
Commits
rCOMMc703b8a03b42: [services] Backup - Send log - finish read request logic
Summary

Depends on D3650

Adding logic for certain state options in order to properly handle sending logs to the backup service.

Test Plan

You need 3 terminals
1:

cd services
yarn run-backup-service

2:

cd services
yarn run-blob-service

3:
You can use https://github.com/karol-bisztyga/grpc-playground/tree/backup-async

./build.sh
./cmake/build/bin/client

then in the GUI use the n option (just type n and hit enter, this stands for creating a new backup).

Diff Detail

Repository
rCOMM Comm
Lint
Lint Not Applicable
Unit
Tests Not Applicable

Event Timeline

• karol created this revision.Apr 7 2022, 7:22 AM
Herald added subscribers: • abosh, • benschac, atul and 3 others. · View Herald TranscriptApr 7 2022, 7:22 AM
• karol added a child revision: D3594: [services] Blob - Fix reading data from S3.Apr 7 2022, 7:23 AM
• karol edited the summary of this revision. (Show Details)Apr 7 2022, 7:33 AM
• karol edited the test plan for this revision. (Show Details)
• karol added reviewers: tomek, • max.
• karol added a parent revision: D3650: [services] Backup - Send log - add terminate callback.
• karol mentioned this in D3586: [services] Backup - Send log - add store in database.Apr 7 2022, 7:36 AM
Harbormaster completed remote builds in B7971: Diff 11175.Apr 7 2022, 8:41 AM
• karol requested review of this revision.Apr 7 2022, 8:41 AM
tomek accepted this revision.Apr 8 2022, 1:56 AM
tomek added inline comments.
services/backup/docker-server/contents/server/src/Reactors/server/SendLogReactor.h
163 ↗(On Diff #11175)

I'm wondering if it was a good idea to store either log or holder in value. Maybe the code would be more readable if we had two separate fields. What do you think?

This revision is now accepted and ready to land.Apr 8 2022, 1:56 AM
• karol updated this revision to Diff 11220.Apr 8 2022, 2:49 AM
• karol edited the summary of this revision. (Show Details)

rebase

Harbormaster completed remote builds in B8015: Diff 11220.Apr 8 2022, 2:55 AM
• karol mentioned this in D3647: [services] Backup - Send log - add backup id and log hash states.Apr 13 2022, 4:19 AM
• karol added inline comments.Apr 13 2022, 4:28 AM
services/backup/docker-server/contents/server/src/Reactors/server/SendLogReactor.h
163 ↗(On Diff #11175)

Well, maybe, Basically, this was agreed on during the arch review. Maybe you're right. If you feel like it, please create a task and assign it to me, we can discuss if it is a good change.

Closed by commit rCOMMc703b8a03b42: [services] Backup - Send log - finish read request logic (authored by • karol). · Explain WhyApr 13 2022, 4:58 AM
This revision was automatically updated to reflect the committed changes.
• karol added a commit: rCOMMc703b8a03b42: [services] Backup - Send log - finish read request logic.
tomek added inline comments.Apr 13 2022, 8:05 AM
services/backup/docker-server/contents/server/src/Reactors/server/SendLogReactor.h
163 ↗(On Diff #11175)

It was just an idea. If you think it would simplify things, we can do this as a followup. If you're not convinced, then creating a task doesn't make a lot of sense. So up to you.

But overall, you've written this code, so do you think that having separate fields would simplify the implementation?

ashoat added inline comments.Apr 18 2022, 11:42 AM
services/backup/docker-server/contents/server/src/Reactors/server/SendLogReactor.h
163 ↗(On Diff #11175)

Basically, this was agreed on during the arch review

It's true we agreed on using the same database field for these two values, but that doesn't mean we need to have the same variable for it in SendLogReactor. I agree with @palys-swm that it would be more readable to use different variables for these

• karol added inline comments.May 6 2022, 4:56 AM
services/backup/docker-server/contents/server/src/Reactors/server/SendLogReactor.h
163 ↗(On Diff #11175)

https://linear.app/comm/issue/ENG-1092/store-holder-and-value-as-2-separate-fields-in-backup

Revision Contents
Changeset List

PathSize
services/
backup/
docker-server/
contents/
server/
src/
Reactors/
server/
41 lines

Diff 11399

View Options

services/backup/docker-server/contents/server/src/Reactors/server/SendLogReactor.h

Show First 20 Lines • Show All 212 Lines • ▼ Show 20 Lines
function reassignMessagesToRealizedThreads( function reassignMessagesToRealizedThreads(
messageStore: MessageStore, messageStore: MessageStore,
threadInfos: RawThreadInfos, threadInfos: RawThreadInfos,
): ReassignmentResult { ): ReassignmentResult {
const pendingToRealizedThreadIDs = const pendingToRealizedThreadIDs =
pendingToRealizedThreadIDsSelector(threadInfos); pendingToRealizedThreadIDsSelector(threadInfos);
const messageStoreOperations: MessageStoreOperation[] = []; const messageStoreOperations: MessageStoreOperation[] = [];
const messages = {}; const messages: { [string]: RawMessageInfo } = {};
for (const storeMessageID in messageStore.messages) { for (const storeMessageID in messageStore.messages) {
const message = messageStore.messages[storeMessageID]; const message = messageStore.messages[storeMessageID];
const newThreadID = pendingToRealizedThreadIDs.get(message.threadID); const newThreadID = pendingToRealizedThreadIDs.get(message.threadID);
messages[storeMessageID] = newThreadID messages[storeMessageID] = newThreadID
? { ? {
...message, ...message,
threadID: newThreadID, threadID: newThreadID,
time: threadInfos[newThreadID]?.creationTime ?? message.time, time: threadInfos[newThreadID]?.creationTime ?? message.time,
} }
: message; : message;
if (!newThreadID) { if (!newThreadID) {
continue; continue;
} }
const updateMsgOperation: ReplaceMessageOperation = { const updateMsgOperation: ReplaceMessageOperation = {
type: 'replace', type: 'replace',
payload: { id: storeMessageID, messageInfo: messages[storeMessageID] }, payload: { id: storeMessageID, messageInfo: messages[storeMessageID] },
}; };
messageStoreOperations.push(updateMsgOperation); messageStoreOperations.push(updateMsgOperation);
} }
const threads = {}; const threads: { [string]: ThreadMessageInfo } = {};
const reassignedThreadIDs = []; const reassignedThreadIDs = [];
const updatedThreads = {}; const updatedThreads: { [string]: ThreadMessageInfo } = {};
const threadsToRemove = []; const threadsToRemove = [];
for (const threadID in messageStore.threads) { for (const threadID in messageStore.threads) {
const threadMessageInfo = messageStore.threads[threadID]; const threadMessageInfo = messageStore.threads[threadID];
const newThreadID = pendingToRealizedThreadIDs.get(threadID); const newThreadID = pendingToRealizedThreadIDs.get(threadID);
if (!newThreadID) { if (!newThreadID) {
threads[threadID] = threadMessageInfo; threads[threadID] = threadMessageInfo;
continue; continue;
} }
▲ Show 20 Lines • Show All 42 Lines • ▼ Show 20 Linestype MergeNewMessagesResult = {
+messageStoreOperations: $ReadOnlyArray<MessageStoreOperation>, +messageStoreOperations: $ReadOnlyArray<MessageStoreOperation>,
+messageStore: MessageStore, +messageStore: MessageStore,
}; };
// oldMessageStore is from the old state // oldMessageStore is from the old state
// newMessageInfos, truncationStatus come from server // newMessageInfos, truncationStatus come from server
function mergeNewMessages( function mergeNewMessages(
oldMessageStore: MessageStore, oldMessageStore: MessageStore,
newMessageInfos: $ReadOnlyArray<RawMessageInfo>, newMessageInfos: $ReadOnlyArray<RawMessageInfo>,
truncationStatus: { [threadID: string]: MessageTruncationStatus }, truncationStatus: { +[threadID: string]: MessageTruncationStatus },
threadInfos: RawThreadInfos, threadInfos: RawThreadInfos,
): MergeNewMessagesResult { ): MergeNewMessagesResult {
const { const {
messageStoreOperations: updateWithLatestThreadInfosOps, messageStoreOperations: updateWithLatestThreadInfosOps,
messageStore: messageStoreUpdatedWithLatestThreadInfos, messageStore: messageStoreUpdatedWithLatestThreadInfos,
reassignedThreadIDs, reassignedThreadIDs,
} = updateMessageStoreWithLatestThreadInfos(oldMessageStore, threadInfos); } = updateMessageStoreWithLatestThreadInfos(oldMessageStore, threadInfos);
▲ Show 20 Lines • Show All 134 Lines • ▼ Show 20 Lines): MergeNewMessagesResult {
)(unshimmedNewMessagesOfWatchedThreads); )(unshimmedNewMessagesOfWatchedThreads);
const newMessageOps: MessageStoreOperation[] = []; const newMessageOps: MessageStoreOperation[] = [];
const threadsToMessageIDs = mapThreadsToMessageIDsFromOrderedMessageInfos( const threadsToMessageIDs = mapThreadsToMessageIDsFromOrderedMessageInfos(
orderedNewMessageInfos, orderedNewMessageInfos,
); );
const oldMessageInfosToCombine = []; const oldMessageInfosToCombine = [];
const threadsThatNeedMessageIDsResorted = []; const threadsThatNeedMessageIDsResorted = [];
const local = {}; const local: { [string]: LocalMessageInfo } = {};
const updatedThreads = {}; const updatedThreads: { [string]: ThreadMessageInfo } = {};
const threads = _flow( const threads = _flow(
_mapValuesWithKeys((messageIDs: string[], threadID: string) => { _mapValuesWithKeys((messageIDs: string[], threadID: string) => {
const oldThread = updatedMessageStore.threads[threadID]; const oldThread = updatedMessageStore.threads[threadID];
const truncate = truncationStatus[threadID]; const truncate = truncationStatus[threadID];
if (!oldThread) { if (!oldThread) {
updatedThreads[threadID] = { updatedThreads[threadID] = {
...newThread(), ...newThread(),
messageIDs, messageIDs,
▲ Show 20 Lines • Show All 167 Lines • ▼ Show 20 Lines): UpdateMessageStoreWithLatestThreadInfosResult {
const { const {
messageStore: reassignedMessageStore, messageStore: reassignedMessageStore,
messageStoreOperations: reassignMessagesOps, messageStoreOperations: reassignMessagesOps,
reassignedThreadIDs, reassignedThreadIDs,
} = reassignMessagesToRealizedThreads(messageStore, threadInfos); } = reassignMessagesToRealizedThreads(messageStore, threadInfos);
messageStoreOperations.push(...reassignMessagesOps); messageStoreOperations.push(...reassignMessagesOps);
const watchedIDs = [...threadWatcher.getWatchedIDs(), ...reassignedThreadIDs]; const watchedIDs = [...threadWatcher.getWatchedIDs(), ...reassignedThreadIDs];
const filteredThreads = {}; const filteredThreads: { [string]: ThreadMessageInfo } = {};
const threadsToRemoveMessagesFrom = []; const threadsToRemoveMessagesFrom = [];
const messageIDsToRemove = []; const messageIDsToRemove = [];
for (const threadID in reassignedMessageStore.threads) { for (const threadID in reassignedMessageStore.threads) {
const threadMessageInfo = reassignedMessageStore.threads[threadID]; const threadMessageInfo = reassignedMessageStore.threads[threadID];
const threadInfo = threadInfos[threadID]; const threadInfo = threadInfos[threadID];
if (isThreadWatched(threadID, threadInfo, watchedIDs)) { if (isThreadWatched(threadID, threadInfo, watchedIDs)) {
filteredThreads[threadID] = threadMessageInfo; filteredThreads[threadID] = threadMessageInfo;
} else { } else {
threadsToRemoveMessagesFrom.push(threadID); threadsToRemoveMessagesFrom.push(threadID);
messageIDsToRemove.push(...threadMessageInfo.messageIDs); messageIDsToRemove.push(...threadMessageInfo.messageIDs);
} }
} }
const updatedThreads = {}; const updatedThreads: { [string]: ThreadMessageInfo } = {};
for (const threadID in threadInfos) { for (const threadID in threadInfos) {
const threadInfo = threadInfos[threadID]; const threadInfo = threadInfos[threadID];
if ( if (
isThreadWatched(threadID, threadInfo, watchedIDs) && isThreadWatched(threadID, threadInfo, watchedIDs) &&
!filteredThreads[threadID] !filteredThreads[threadID]
) { ) {
filteredThreads[threadID] = newThread(); filteredThreads[threadID] = newThread();
updatedThreads[threadID] = filteredThreads[threadID]; updatedThreads[threadID] = filteredThreads[threadID];
▲ Show 20 Lines • Show All 195 Lines • ▼ Show 20 Lines return mergeNewMessages(
[action.payload.messageInfo], [action.payload.messageInfo],
{ {
[action.payload.messageInfo.threadID]: [action.payload.messageInfo.threadID]:
messageTruncationStatus.UNCHANGED, messageTruncationStatus.UNCHANGED,
}, },
newThreadInfos, newThreadInfos,
); );
} else if (action.type === registerActionTypes.success) { } else if (action.type === registerActionTypes.success) {
const truncationStatuses = {}; const truncationStatuses: { [string]: MessageTruncationStatus } = {};
for (const messageInfo of action.payload.rawMessageInfos) { for (const messageInfo of action.payload.rawMessageInfos) {
truncationStatuses[messageInfo.threadID] = truncationStatuses[messageInfo.threadID] =
messageTruncationStatus.EXHAUSTIVE; messageTruncationStatus.EXHAUSTIVE;
} }
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
action.payload.rawMessageInfos, action.payload.rawMessageInfos,
truncationStatuses, truncationStatuses,
Show All 32 Lines ) {
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
messagesResult.rawMessageInfos, messagesResult.rawMessageInfos,
messagesResult.truncationStatuses, messagesResult.truncationStatuses,
newThreadInfos, newThreadInfos,
); );
} else if (action.type === sendEditMessageActionTypes.success) { } else if (action.type === sendEditMessageActionTypes.success) {
const { newMessageInfos } = action.payload; const { newMessageInfos } = action.payload;
const truncationStatuses = {}; const truncationStatuses: { [string]: MessageTruncationStatus } = {};
for (const messageInfo of newMessageInfos) { for (const messageInfo of newMessageInfos) {
truncationStatuses[messageInfo.threadID] = truncationStatuses[messageInfo.threadID] =
messageTruncationStatus.UNCHANGED; messageTruncationStatus.UNCHANGED;
} }
return mergeNewMessages( return mergeNewMessages(
messageStore, messageStore,
newMessageInfos, newMessageInfos,
truncationStatuses, truncationStatuses,
▲ Show 20 Lines • Show All 210 Lines • ▼ Show 20 Lines return {
messageStore: { messageStore: {
...messageStore, ...messageStore,
messages: processedMessageStore.messages, messages: processedMessageStore.messages,
threads: processedMessageStore.threads, threads: processedMessageStore.threads,
local, local,
}, },
}; };
} else if (action.type === saveMessagesActionType) { } else if (action.type === saveMessagesActionType) {
const truncationStatuses = {}; const truncationStatuses: { [string]: MessageTruncationStatus } = {};
for (const messageInfo of action.payload.rawMessageInfos) { for (const messageInfo of action.payload.rawMessageInfos) {
truncationStatuses[messageInfo.threadID] = truncationStatuses[messageInfo.threadID] =
messageTruncationStatus.UNCHANGED; messageTruncationStatus.UNCHANGED;
} }
const { messageStoreOperations, messageStore: newMessageStore } = const { messageStoreOperations, messageStore: newMessageStore } =
mergeNewMessages( mergeNewMessages(
messageStore, messageStore,
action.payload.rawMessageInfos, action.payload.rawMessageInfos,
Show All 10 Lines return {
// RawMessageInfo, so we want to make sure we still fetch it // RawMessageInfo, so we want to make sure we still fetch it
currentAsOf: messageStore.currentAsOf, currentAsOf: messageStore.currentAsOf,
}, },
}; };
} else if (action.type === messageStorePruneActionType) { } else if (action.type === messageStorePruneActionType) {
const now = Date.now(); const now = Date.now();
const messageIDsToPrune = []; const messageIDsToPrune = [];
const updatedThreads = {}; const updatedThreads: { [string]: ThreadMessageInfo } = {};
for (const threadID of action.payload.threadIDs) { for (const threadID of action.payload.threadIDs) {
let thread = messageStore.threads[threadID]; let thread = messageStore.threads[threadID];
if (!thread) { if (!thread) {
continue; continue;
} }
thread = { ...thread, lastPruned: now }; thread = { ...thread, lastPruned: now };
const newMessageIDs = [...thread.messageIDs]; const newMessageIDs = [...thread.messageIDs];
▲ Show 20 Lines • Show All 355 Lines • ▼ Show 20 Lines ) {
} }
const threadsNeedMsgIDsResorting = new Set(); const threadsNeedMsgIDsResorting = new Set();
const actionPayloadMessages = const actionPayloadMessages =
messageStoreOpsHandlers.translateClientDBData(payloadMessages); messageStoreOpsHandlers.translateClientDBData(payloadMessages);
// When starting the app on native, we filter out any local-only multimedia // When starting the app on native, we filter out any local-only multimedia
// messages because the relevant context is no longer available // messages because the relevant context is no longer available
const messageIDsToBeRemoved = []; const messageIDsToBeRemoved = [];
const threadsToAdd = {}; const threadsToAdd: { [string]: ThreadMessageInfo } = {};
for (const id in actionPayloadMessages) { for (const id in actionPayloadMessages) {
const message = actionPayloadMessages[id]; const message = actionPayloadMessages[id];
const { threadID } = message; const { threadID } = message;
let existingThread = threads[threadID]; let existingThread = threads[threadID];
if (!existingThread) { if (!existingThread) {
existingThread = newThread(); existingThread = newThread();
threadsToAdd[threadID] = existingThread; threadsToAdd[threadID] = existingThread;
} }
▲ Show 20 Lines • Show All 104 Lines • Show Last 20 Lines