opProcessingController.ts

/*!
 * Copyright (c) Microsoft Corporation. All rights reserved.
 * Licensed under the MIT License.
 */

import { strict as assert } from "assert";
import { IDeltaManager } from "@fluidframework/container-definitions";
import {
    IDocumentMessage,
    ISequencedDocumentMessage,
    ISequencedDocumentSystemMessage,
    MessageType,
} from "@fluidframework/protocol-definitions";
import { debug } from "./debug";

// An IDeltaManager alias to be used within this class.
export type DeltaManager = IDeltaManager<ISequencedDocumentMessage, IDocumentMessage>;

class DeltaManagerToggle {
    private inboundPauseP: Promise<void> | undefined;
    private outboundPauseP: Promise<void> | undefined;
    constructor(public readonly deltaManager: DeltaManager) {
    }

    public async togglePauseAll() {
        return Promise.all([this.togglePauseInbound(), this.togglePauseOutbound()]);
    }

    public toggleResumeAll() {
        this.toggleResumeInbound();
        this.toggleResumeOutbound();
    }
    public async togglePauseInbound() {
        if (!this.inboundPauseP) {
            this.inboundPauseP = this.deltaManager.inbound.pause();
        }
        return this.inboundPauseP;
    }

    public async togglePauseOutbound() {
        if (!this.outboundPauseP) {
            this.outboundPauseP = this.deltaManager.outbound.pause();
        }
        return this.outboundPauseP;
    }

    public toggleResumeInbound() {
        if (this.inboundPauseP) {
            this.inboundPauseP = undefined;
            this.deltaManager.inbound.resume();
        }
    }

    public toggleResumeOutbound() {
        if (this.outboundPauseP) {
            this.outboundPauseP = undefined;
            this.deltaManager.outbound.resume();
        }
    }

    public get inboundPaused() {
        return this.inboundPauseP !== undefined;
    }
}

/**
 * Monitor for DeltaManager, and track in/out ops to figure out whether there are
 * outstanding ops that the server hasn't ack yet.  Used by the OpProcessingController
 * to wait for all the ops has round tripped.
 *
 * For outbound, we monitor ops leaving the outbound queue on the "op" event.
 * For inbound, we monitor the first moment we see an op coming back on the "push" event.
 *
 * It also monitor connect and disconnect state so that we can refresh the tracking and clientId
 *
 * The monitor ignores ops generated by the server. It also don't track NoOp since the server
 * might coalesce them with other ops, or a single NoOp, or delay it if it don't think it is necessary
 */
class DeltaManagerMonitor extends DeltaManagerToggle {
    private pendingCount: number = 0;
    public clientId: string | undefined;
    public readMode = true;
    private firstClientSequenceNumber: number = -1;
    private lastOutbound: IDocumentMessage | undefined;
    private readonly pendingLeaveClientIds = new Set<string>();
    private readonly lastInboundPerClient = new Map<string, ISequencedDocumentMessage>();
    private pendingWriteConnection = false;

    /**
     * Determines if this monitor should expect work/ops from the outbound monitor.
     * @param outbound - the monitor who's outbound to consider
     */
    public expectingInboundFrom(outbound: DeltaManagerMonitor): boolean {
        // there should be no outstanding work for disposed delta managers
        if (this.deltaManager.disposed || outbound.deltaManager.disposed) {
            return false;
        }
        // if there is no last outbound, we are not waiting for anything
        if (outbound.lastOutbound === undefined
            || outbound.clientId === undefined) {
            return false;
        }
        // if out inbound is paused we are not expecting to receive anything more
        if (this.inboundPaused) {
            return false;
        }

        // if outbound is ourself, return if we having pending work
        if (this === outbound) {
            return this.hasPendingWork();
        }

        // check if we are waiting to see a message from outbound
        const lastInboundForOutbound = this.lastInboundPerClient.get(outbound.clientId);
        if (lastInboundForOutbound !== undefined) {
            return outbound.lastOutbound.clientSequenceNumber > lastInboundForOutbound.clientSequenceNumber;
        }

        // has pending work will be true for outbound until it receives it's own seq
        // this check ensures the other client has seen the same ops as the outbound
        return outbound.latestSequenceNumber > this.latestSequenceNumber;
    }

    constructor(deltaManager: DeltaManager) {
        super(deltaManager);

        // The deltaManager may be connected already, need to get the clientId.
        // TODO: hackery to get the clientId from the delta manager, find a better way
        const anyDeltaManager = deltaManager as any;
        // Unwrap the proxy if there is any
        const fullDeltaManager = (anyDeltaManager.deltaManager ?? anyDeltaManager);
        const id = fullDeltaManager.connection?.clientId;
        if (id !== undefined) {
            this.connect(id);
        }

        deltaManager.on("connect", (details) => this.connect(details.clientId));
        deltaManager.on("disconnect", (reason) => {
            assert(this.clientId !== undefined);
            this.trace("DIS");
            this.clientId = undefined;
            // Once disconnected, the runtime is going to keep track of ops and replay as necessary
            // Clear the pending count and start anew
            this.pendingCount = 0;
            this.firstClientSequenceNumber = -1;
            this.lastOutbound = undefined;
        });
        deltaManager.outbound.on("op", this.outbound.bind(this));
        deltaManager.inbound.on("push", this.inbound.bind(this));
    }

    public get latestSequenceNumber() {
        return this.deltaManager.lastSequenceNumber;
    }

    public hasPendingWork() {
        return !this.deltaManager.disposed
            && (this.pendingWriteConnection || this.pendingCount !== 0 || this.pendingLeaveClientIds.size !== 0);
    }

    private connect(clientId: string) {
        this.clientId = clientId;
        this.readMode = !this.deltaManager.active;
        this.trace("CON");
    }
    private inbound(message: ISequencedDocumentMessage) {
        if (message.clientId) {
            this.lastInboundPerClient.set(message.clientId, message);
        }
        if (message.type === MessageType.ClientLeave) {
            const systemLeaveMessage = message as ISequencedDocumentSystemMessage;
            const clientId = JSON.parse(systemLeaveMessage.data) as string;
            this.lastInboundPerClient.delete(clientId);
            this.pendingLeaveClientIds.delete(clientId);
        }

        if (this.clientId === undefined) {
            // Ignore message when we are not connected.
            return;
        }

        if (message.clientId === undefined || message.clientId !== this.clientId) {
            this.trace("SEQ", message.type);
            return;
        }

        if (this.firstClientSequenceNumber === -1 || this.firstClientSequenceNumber > message.clientSequenceNumber) {
            this.trace("SEQ", message.type);
            // if we haven't seen any outbound or the message is before the outbound message that we have seen,
            // then message is sent before we start monitoring, ignore.
            return;
        }

        // Need to filter system messages
        switch (message.type) {
            case MessageType.ClientJoin:
            case MessageType.ClientLeave:
                assert(false, "join and leave message shouldn't have clientId");
            // These are generated by the server, don't count
            case MessageType.NoOp:
            case MessageType.NoClient:
                this.trace("SEQ", message.type);
                break;
            default:
                assert(this.pendingCount);
                this.pendingCount--;
                this.trace("IN", message.type);
        }
    }

    private outbound(messages: IDocumentMessage[]) {
        assert(this.clientId);
        assert(messages.length);
        if (this.firstClientSequenceNumber === -1) {
            // save the client sequence number of the first outbound message we see
            // to exclude any message that was sent before we start monitoring the delta manager
            this.firstClientSequenceNumber = messages[0].clientSequenceNumber;
        }
        // if we are not active, the outbound with nack, and we will reconnect write
        // this flag tracks the process. after reconnection, the op will be resubmitted
        // on the write connection and reset this flag
        this.pendingWriteConnection = !this.deltaManager.active;
        for (const message of messages) {
            // No-op's are not directly broadcast
            // the server coaleses and send it's own
            // no-op if no user messages arrive
            // to bump min seq
            if (message.type !== MessageType.NoOp) {
                this.pendingCount++;
                this.lastOutbound = message;
            }
            this.trace("OUT", message.type);
        }
    }

    public trace(action: string, op?: string) {
        debug(`DeltaConnectionMonitor: ${action.padEnd(3)}: ${this.clientId} `
            + `pending:${this.pendingCount} seq:${this.latestSequenceNumber} ${op ?? ""}`);
    }

    public onClientDisconnect(clientId: string) {
        // Keep track of a list of clientIds that we expect leave message from
        this.pendingLeaveClientIds.add(clientId);
    }
}
/**
 * @deprecated OpProcessingController has been improved to not need server information and work against other servers.
 *      So this is no longer necessary, and allows this and test to be run against different endpoints.
 */
export interface IDeltaConnectionServerMonitor {
    hasPendingWork(): Promise<boolean>;
}

/**
 * Class with access to the local delta connection server and delta managers that can control op processing.
 *
 * @deprecated Can be removed \>=0.38. Replaced with LoaderContainerTracker
 */
export class OpProcessingController {
    /**
     * Yields control in the JavaScript event loop.
     */
    public static async yield(): Promise<void> {
        await new Promise<void>((resolve) => {
            setTimeout(resolve, 0);
        });
    }

    private readonly deltaManagerMonitors = new Map<DeltaManager, DeltaManagerMonitor>();

    private isNormalProcessingPaused = false;

    /*
    * Is processing being deterministically controlled, or are changes allowed to flow freely?
    */
    public get isProcessingControlled(): boolean {
        return this.isNormalProcessingPaused;
    }

    /**
     * @param deltaConnectionServerMonitor - delta connection server monitor to tell whether we have
     *  pending work
     */
    public constructor(private readonly deltaConnectionServerMonitor?: IDeltaConnectionServerMonitor) { }

    /**
     * Add a collection of delta managers by adding them to the local collection.
     * @param deltaManagers - Array of deltaManagers to add
     */
    public addDeltaManagers(...deltaManagers: DeltaManager[]) {
        deltaManagers.forEach((deltaManager) => {
            const monitorSetup = (monitor1: DeltaManagerMonitor, monitor2: DeltaManagerMonitor) => {
                if (monitor1.clientId !== undefined && monitor1.deltaManager.active) {
                    const clientId = monitor1.clientId;
                    monitor1.deltaManager.once("disconnect", () => {
                        monitor2.onClientDisconnect(clientId);
                    });
                }
                monitor1.deltaManager.on("connect", (details) => {
                    if (monitor1.deltaManager.active) {
                        monitor1.deltaManager.once("disconnect", () => {
                            monitor2.onClientDisconnect(details.clientId);
                        });
                    }
                });
            };

            // Wire up event listener so we can keep track of leave message that we expects
            const newMonitor = new DeltaManagerMonitor(deltaManager);
            for (const monitor of this.deltaManagerMonitors.values()) {
                monitorSetup(newMonitor, monitor);
                monitorSetup(monitor, newMonitor);
            }

            this.deltaManagerMonitors.set(deltaManager, newMonitor);
        });
    }

    /**
      * Processes incoming and outgoing op) of the given delta managers.
      * It validates the delta managers and resumes its inbound and outbound queues. It then keeps yielding
      * the JS event loop until all the ops have been processed by the server and by the delta managers.
      *
      * @param deltaMangers - Array of delta managers whose ops to process. If no delta manager is provided, it
      * processes the ops for all the delta managers in our collection.
      */
    public async process(...deltaMangers: DeltaManager[]): Promise<void> {
        const monitors = this.mapDeltaManagerMonitor(deltaMangers);

        // Pause the queues of all the delta managers in our collection to make sure that we only process the ops of
        // the requested delta managers.
        await this.pauseAllDeltaManagerQueues();

        // Resume the delta queues so that we can process incoming and outgoing ops.
        monitors.forEach((monitor) => monitor.toggleResumeAll());

        // Wait for all pending ops to be processed.
        await this.yieldWhileDeltaManagersHaveWork(
            monitors,
            (deltaManager) => !deltaManager.inbound.idle || !deltaManager.outbound.idle);
    }

    /**
     * Processes incoming ops of the given delta managers.
     * It validates the delta managers and resumes its inbound queue. It then keeps yielding the JS event loop until
     * all the ops have been processed by the server and by the delta managers.
     *
     * @param deltaMangers - Array of delta managers whose incoming ops to process. If no delta manager is provided, it
     * processes the ops for all the delta managers in our collection.
     */
    public async processIncoming(...deltaMangers: DeltaManager[]): Promise<void> {
        const monitors = this.mapDeltaManagerMonitor(deltaMangers);

        // Pause the queues of all the delta managers in our collection to make sure that we only process the incoming
        // ops of the requested delta managers.
        await this.pauseAllDeltaManagerQueues();

        // Resume the inbound delta queue so that we can process incoming ops.
        monitors.forEach((monitor) => {
            monitor.toggleResumeInbound();
        });

        // Wait for all pending incoming ops to be processed.
        await this.yieldWhileDeltaManagersHaveWork(
            monitors,
            (deltaManager) => !deltaManager.inbound.idle);
    }

    /**
     * Processes outgoing ops of the given delta managers.
     * It validates the delta managers and resumes its outbound queue. It then keeps yielding the JS event loop until
     * all the ops have been processed by the server and by the delta managers.
     *
     * @param deltaMangers - Array of delta managers whose outgoing ops to process. If no delta manager is provided, it
     * processes the ops for all the delta managers in our collection.
     */
    public async processOutgoing(...deltaMangers: DeltaManager[]): Promise<void> {
        const monitors = this.mapDeltaManagerMonitor(deltaMangers);

        // Pause the queues of all the delta managers in our collection to make sure that we only process the outgoing
        // ops of the requested delta managers.
        await this.pauseAllDeltaManagerQueues();

        // Resume the outbound delta queue so that we can process outgoing ops.
        monitors.forEach((monitor) => {
            monitor.toggleResumeOutbound();
        });

        // Wait for all pending outgoing ops to be processed.
        await this.yieldWhileDeltaManagersHaveWork(
            monitors,
            (deltaManager) => !deltaManager.outbound.idle);
    }

    /**
     * Pauses the delta processing for controlled testing by pausing the inbound and outbound queues of the delta
     * managers.
     *
     * @param deltaMangers - Array of delta managers whose processing to pause. If no delta manager is provided, it
     * pauses the processing of all the delta managers in our collection.
     */
    public async pauseProcessing(...deltaMangers: DeltaManager[]) {
        const monitors = this.mapDeltaManagerMonitor(deltaMangers);

        // Pause the inbound and outbound delta queues.
        await this.pauseDeltaManagerQueues(monitors);

        this.isNormalProcessingPaused = true;
    }

    /**
     * Resumes the delta processing after a pauseProcessing calls by resuming the inbound and outbound queues of
     * the delta managers.
     *
     * @param deltaMangers - Array of delta managers whose processing to resume. If no delta manager is provided, it
     * resumes the processing of all the delta managers in our collection.
     */
    public resumeProcessing(...deltaMangers: DeltaManager[]) {
        const monitors = this.mapDeltaManagerMonitor(deltaMangers);

        // Resume the inbound and outbound delta queues.
        monitors.forEach((monitor) => monitor.toggleResumeAll());

        this.isNormalProcessingPaused = false;
    }

    /**
     * Map a list of DeltaManager to its monitor.  Throw an error if the delta manager is not in our collection
     * @param deltaMangers - The delta managers to get the monitors for
     */
    private mapDeltaManagerMonitor(deltaMangers: DeltaManager[]) {
        if (deltaMangers.length === 0) {
            // If no delta managers are provided, process all delta managers in our collection.
            return Array.from(this.deltaManagerMonitors.values());
        }

        return deltaMangers.map((deltaManager) => {
            const monitor = this.deltaManagerMonitors.get(deltaManager);
            assert(monitor, "All delta managers must be added to deterministically control processing");
            return monitor;
        });
    }

    /**
     * It keeps yielding the JS event loop until  all the ops have been processed by the server and by the passed
     * delta managers.
     * @param monitors - The delta managers should ops have to be processed.
     * @param hasWork - Function that tells if the delta manager has pending work or not.
     */
    private async yieldWhileDeltaManagersHaveWork(
        monitors: Iterable<DeltaManagerMonitor>,
        hasWork: (deltaManagers: DeltaManager) => boolean,
    ): Promise<void> {
        let working: boolean;
        do {
            await OpProcessingController.yield();
            working = false;
            if (await this.deltaConnectionServerMonitor?.hasPendingWork() === true) {
                working = true;
            } else {
                for (const monitor of monitors) {
                    if (!monitor.deltaManager.disposed) {
                        if (monitor.hasPendingWork() || hasWork(monitor.deltaManager)) {
                            working = true;
                            break;
                        }

                        for (const outBoundMonitor of monitors) {
                            if (monitor !== outBoundMonitor) {
                                if (monitor.expectingInboundFrom(outBoundMonitor)) {
                                    working = true;
                                    break;
                                }
                            }
                        }
                        if (working === true) {
                            break;
                        }
                    }
                }
            }
        } while (working);

        // If deterministically controlling events, need to pause before continuing
        if (this.isNormalProcessingPaused) {
            await this.pauseDeltaManagerQueues(monitors);
        }
    }

    /**
     * Pauses the inbound and outbound queues of all the delta managers given
     * @param monitors - The delta managers should ops have to be processed.
     */
    private async pauseDeltaManagerQueues(monitors: Iterable<DeltaManagerToggle>) {
        const p: Promise<[void, void]>[] = [];
        for (const monitor of monitors) {
            p.push(monitor.togglePauseAll());
        }
        return Promise.all(p);
    }

    /**
     * Pauses the inbound and outbound queues of all the delta managers in our collection.
     */
    private async pauseAllDeltaManagerQueues() {
        return this.pauseDeltaManagerQueues(this.deltaManagerMonitors.values());
    }
}