Fetching ops from PUSH

packages/drivers/driver-base/package.json

@@ -34,6 +34,7 @@
     "@fluidframework/driver-definitions": "^0.39.6",
     "@fluidframework/driver-utils": "^0.45.0",
     "@fluidframework/protocol-definitions": "^0.1024.0",
+    "@fluidframework/telemetry-utils": "^0.45.0",
     "debug": "^4.1.1"
   },
   "devDependencies": {

packages/drivers/driver-base/src/documentDeltaConnection.ts

@@ -23,6 +23,7 @@ import {
     ScopeType,
 } from "@fluidframework/protocol-definitions";
 import { IDisposable, ITelemetryLogger } from "@fluidframework/common-definitions";
+import { ChildLogger } from "@fluidframework/telemetry-utils";
 import { debug } from "./debug";
 
 /**
@@ -71,6 +72,7 @@ export class DocumentDeltaConnection
      * After disconnection, we flip this to prevent any stale messages from being emitted.
      */
     protected _disposed: boolean = false;
+    protected readonly logger: ITelemetryLogger;
 
     public get details(): IConnected {
         if (!this._details) {
@@ -86,10 +88,12 @@ export class DocumentDeltaConnection
     protected constructor(
         protected readonly socket: SocketIOClient.Socket,
         public documentId: string,
-        protected readonly logger: ITelemetryLogger,
+        logger: ITelemetryLogger,
     ) {
         super();
 
+        this.logger = ChildLogger.create(logger, "DeltaConnection");
+
         this.submitManager = new BatchManager<IDocumentMessage[]>(
             (submitType, work) => {
                 // Although the implementation here disconnects the socket and does not reuse it, other subclasses

packages/drivers/odsp-driver/src/odspDocumentDeltaConnection.ts

@@ -6,7 +6,7 @@
 import { ITelemetryLogger } from "@fluidframework/common-definitions";
 import { assert } from "@fluidframework/common-utils";
 import { DocumentDeltaConnection } from "@fluidframework/driver-base";
-import { IDocumentDeltaConnection, DriverError } from "@fluidframework/driver-definitions";
+import { DriverError } from "@fluidframework/driver-definitions";
 import {
     IClient,
     IConnect,
@@ -182,7 +182,7 @@ export class OdspDocumentDeltaConnection extends DocumentDeltaConnection {
         url: string,
         telemetryLogger: ITelemetryLogger,
         timeoutMs: number,
-        epochTracker: EpochTracker): Promise<IDocumentDeltaConnection>
+        epochTracker: EpochTracker): Promise<OdspDocumentDeltaConnection>
     {
         // enable multiplexing when the websocket url does not include the tenant/document id
         const parsedUrl = new URL(url);
@@ -244,6 +244,7 @@ export class OdspDocumentDeltaConnection extends DocumentDeltaConnection {
     }
 
     private socketReference: SocketReference | undefined;
+    private readonly requestOpsNoncePrefix: string;
 
     /**
      * Error raising for socket.io issues
@@ -309,6 +310,15 @@ export class OdspDocumentDeltaConnection extends DocumentDeltaConnection {
     {
         super(socket, documentId, logger);
         this.socketReference = socketReference;
+        this.requestOpsNoncePrefix = `${this.documentId}-`;
+    }
+
+    public requestOps(from: number, to: number) {
+        this.socket.emit("get_ops", this.clientId, {
+            nonce: `${this.requestOpsNoncePrefix}${uuid()}`,
+            from,
+            to,
+        });
     }
 
     protected async initialize(connectMessage: IConnect, timeout: number) {
@@ -321,6 +331,20 @@ export class OdspDocumentDeltaConnection extends DocumentDeltaConnection {
             };
         }
 
+        this.socket.on("get_ops_response", (result) => {
+            const messages = result.messages as ISequencedDocumentMessage[];
+            if ((result.nonce as string).startsWith(this.requestOpsNoncePrefix) && messages.length > 0) {
+                this.logger.sendTelemetryEvent({
+                    eventName: "GetOps",
+                    first: messages[0].sequenceNumber,
+                    last: messages[messages.length - 1].sequenceNumber,
+                    code: result.code,
+                    length: messages.length,
+                });
+                this.socket.emit("op", this.documentId, messages);
+            }
+        });
+
         return super.initialize(connectMessage, timeout);
     }
 

packages/drivers/odsp-driver/src/odspDocumentService.ts

@@ -88,6 +88,8 @@ export class OdspDocumentService implements IDocumentService {
 
     private _opsCache?: OpsCache;
 
+    private currentConnection?: OdspDocumentDeltaConnection;
+
     /**
      * @param odspResolvedUrl - resolved url identifying document that will be managed by this service instance.
      * @param getStorageToken - function that can provide the storage token. This is is also referred to as
@@ -184,6 +186,9 @@ export class OdspDocumentService implements IDocumentService {
             concurrency,
             async (from, to, telemetryProps) => service.get(from, to, telemetryProps),
             async (from, to) => {
+                if (this.currentConnection !== undefined && !this.currentConnection.disposed) {
+                    this.currentConnection.requestOps(from, to);
+                }
                 const res = await this.opsCache?.get(from, to);
                 return res as ISequencedDocumentMessage[] ?? [];
             },
@@ -247,6 +252,7 @@ export class OdspDocumentService implements IDocumentService {
                 connection.on("op", (documentId, ops: ISequencedDocumentMessage[]) => {
                     this.opsReceived(ops);
                 });
+                this.currentConnection = connection;
                 return connection;
             } catch (error) {
                 this.cache.sessionJoinCache.remove(this.joinSessionKey);
@@ -294,7 +300,7 @@ export class OdspDocumentService implements IDocumentService {
         io: SocketIOClientStatic,
         client: IClient,
         webSocketUrl: string,
-    ): Promise<IDocumentDeltaConnection> {
+    ): Promise<OdspDocumentDeltaConnection> {
         const startTime = performance.now();
         const connection = await OdspDocumentDeltaConnection.create(
             tenantId,