fix: read time should be used for transaction reads

src/request.ts

@@ -44,7 +44,7 @@
   KeyProto,
   ResponseResult,
   Entities,
   ValueProto,
 } from './entity';
 import {
   Query,
@@ -609,9 +609,9 @@
           const results = res.batch.aggregationResults;
           const finalResults = results
             .map(
               (aggregationResult: any) => aggregationResult.aggregateProperties
             )
             .map((aggregateProperties: any) =>
               Object.fromEntries(
                 new Map(
                   Object.keys(aggregateProperties).map(key => [
@@ -946,7 +946,7 @@
     callback?: SaveCallback
   ): void | Promise<CommitResponse> {
     const transaction = this.datastore.transaction();
     transaction.run(async (err: any) => {
       if (err) {
         try {
           await transaction.rollback();
@@ -1025,9 +1025,13 @@
         );
       }
 
-      reqOpts.readOptions = {
-        transaction: this.id,
-      };
+      if (reqOpts.readOptions) {
+        Object.assign(reqOpts.readOptions, {transaction: this.id});
+      } else {
+        reqOpts.readOptions = {
+          transaction: this.id,
+        };
+      }
     }
 
     datastore.auth.getProjectId((err, projectId) => {

system-test/datastore.ts

@@ -1931,6 +1931,29 @@ async.each(
         });
       });
       describe('transactions', () => {
+        before(async () => {
+          // This 'sleep' function is used to ensure that when data is saved to datastore,
+          // the time on the server is far enough ahead to be sure to be later than timeBeforeDataCreation
+          // so that when we read at timeBeforeDataCreation we get a snapshot of data before the save.
+          const key = datastore.key(['Company', 'Google']);
+          function sleep(ms: number) {
+            return new Promise(resolve => setTimeout(resolve, ms));
+          }
+          // Save for a key so that a read time can be accessed for snapshot reads.
+          const emptyData = {
+            key,
+            data: {},
+          };
+          await datastore.save(emptyData);
+          // Sleep for 10 seconds to ensure timeBeforeDataCreation includes the empty data
+          await sleep(10000);
+          timeBeforeDataCreation = await getReadTime([
+            {kind: 'Company', name: 'Google'},
+          ]);
+          // Sleep for 10 seconds so that any future reads will be later than timeBeforeDataCreation.
+          await sleep(10000);
+        });
+
         it('should run in a transaction', async () => {
           const key = datastore.key(['Company', 'Google']);
           const obj = {
@@ -2031,6 +2054,25 @@ async.each(
           await transaction.commit();
         });
 
+        it('should query within a transaction at a previous read time', async () => {
+          const transaction = datastore.transaction();
+          await transaction.run();
+          const query = transaction.createQuery('Company');
+          let entitiesBefore;
+          let entitiesNow;
+          try {
+            [entitiesBefore] = await query.run({
+              readTime: timeBeforeDataCreation,
+            });
+            [entitiesNow] = await query.run({});
+          } catch (e) {
+            await transaction.rollback();
+            return;
+          }
+          assert(entitiesBefore!.length < entitiesNow!.length);
+          await transaction.commit();
+        });
+
         describe('aggregate query within a transaction', async () => {
           it('should run a query and return the results', async () => {
             // Add a test here to verify what the data is at this time.

test/gapic-mocks/runQuery.ts

@@ -0,0 +1,118 @@
+// Copyright 2024 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import * as assert from 'assert';
+import {describe} from 'mocha';
+import {DatastoreClient, Datastore} from '../../src';
+import * as protos from '../../protos/protos';
+import {Callback} from 'google-gax';
+
+describe('Run Query', () => {
+  const PROJECT_ID = 'project-id';
+  const NAMESPACE = 'namespace';
+  const clientName = 'DatastoreClient';
+  const options = {
+    projectId: PROJECT_ID,
+    namespace: NAMESPACE,
+  };
+  const datastore = new Datastore(options);
+  // By default, datastore.clients_ is an empty map.
+  // To mock out commit we need the map to contain the Gapic data client.
+  // Normally a call to the data client through the datastore object would initialize it.
+  // We don't want to make this call because it would make a grpc request.
+  // So we just add the data client to the map.
+  const gapic = Object.freeze({
+    v1: require('../../src/v1'),
+  });
+  datastore.clients_.set(clientName, new gapic.v1[clientName](options));
+
+  // This function is used for doing assertion checks.
+  // The idea is to check that the right request gets passed to the commit function in the Gapic layer.
+  function setRunQueryComparison(
+    compareFn: (request: protos.google.datastore.v1.IRunQueryRequest) => void
+  ) {
+    const dataClient = datastore.clients_.get(clientName);
+    if (dataClient) {
+      dataClient.runQuery = (
+        request: any,
+        options: any,
+        callback: (
+          err?: unknown,
+          res?: protos.google.datastore.v1.IRunQueryResponse
+        ) => void
+      ) => {
+        try {
+          compareFn(request);
+        } catch (e) {
+          callback(e);
+        }
+        callback(null, {
+          batch: {
+            moreResults:
+              protos.google.datastore.v1.QueryResultBatch.MoreResultsType
+                .NO_MORE_RESULTS,
+          },
+        });
+      };
+    }
+  }
+
+  it('should pass read time into runQuery for transactions', async () => {
+    // First mock out beginTransaction
+    const dataClient = datastore.clients_.get(clientName);
+    const testId = Buffer.from(Array.from(Array(100).keys()));
+    if (dataClient) {
+      dataClient.beginTransaction = (
+        request: protos.google.datastore.v1.IBeginTransactionRequest,
+        options: protos.google.datastore.v1.IBeginTransactionResponse,
+        callback: Callback<
+          protos.google.datastore.v1.IBeginTransactionResponse,
+          | protos.google.datastore.v1.IBeginTransactionRequest
+          | null
+          | undefined,
+          {} | null | undefined
+        >
+      ) => {
+        callback(null, {
+          transaction: testId,
+        });
+      };
+    }
+    setRunQueryComparison(
+      (request: protos.google.datastore.v1.IRunQueryRequest) => {
+        assert.deepStrictEqual(request, {
+          readOptions: {
+            transaction: testId,
+            readTime: {
+              seconds: 77,
+            },
+          },
+          partitionId: {
+            namespaceId: 'namespace',
+          },
+          query: {
+            distinctOn: [],
+            kind: [{name: 'Task'}],
+            order: [],
+            projection: [],
+          },
+          projectId: 'project-id',
+        });
+      }
+    );
+    const transaction = datastore.transaction();
+    const query = datastore.createQuery('Task');
+    await transaction.runQuery(query, {readTime: 77000});
+  });
+});