Skip to content

Commit

Permalink
Revert Database.runTransactionAsync changes to allow a spin-off PR
Browse files Browse the repository at this point in the history
  • Loading branch information
odeke-em committed Oct 21, 2024
1 parent 3db6a9c commit 13bc9f5
Show file tree
Hide file tree
Showing 2 changed files with 77 additions and 85 deletions.
81 changes: 40 additions & 41 deletions observability-test/spanner.ts
Original file line number Diff line number Diff line change
Expand Up @@ -427,53 +427,55 @@ describe('EndToEnd', () => {
);
});

it('runTransaction', async () => {
it('runTransaction', done => {
const withAllSpansHaveDBName = generateWithAllSpansHaveDBName(
database.formattedName_
);

await database.runTransactionAsync(async transaction => {
return await transaction!.run('SELECT 1');
});
database.runTransaction(async (err, transaction) => {
assert.ifError(err);
const [rows] = await transaction!.run('SELECT 1');
await transaction!.commit();

traceExporter.forceFlush();
const spans = traceExporter.getFinishedSpans();
withAllSpansHaveDBName(spans);
traceExporter.forceFlush();
const spans = traceExporter.getFinishedSpans();
withAllSpansHaveDBName(spans);

const actualEventNames: string[] = [];
const actualSpanNames: string[] = [];
spans.forEach(span => {
actualSpanNames.push(span.name);
span.events.forEach(event => {
actualEventNames.push(event.name);
const actualEventNames: string[] = [];
const actualSpanNames: string[] = [];
spans.forEach(span => {
actualSpanNames.push(span.name);
span.events.forEach(event => {
actualEventNames.push(event.name);
});
});
});

const expectedSpanNames = [
'CloudSpanner.Snapshot.runStream',
'CloudSpanner.Snapshot.run',
'CloudSpanner.Database.runTransactionAsync',
];
assert.deepStrictEqual(
actualSpanNames,
expectedSpanNames,
`span names mismatch:\n\tGot: ${actualSpanNames}\n\tWant: ${expectedSpanNames}`
);
const expectedSpanNames = [
'CloudSpanner.Snapshot.runStream',
'CloudSpanner.Snapshot.run',
'CloudSpanner.Database.runTransaction',
];
assert.deepStrictEqual(
actualSpanNames,
expectedSpanNames,
`span names mismatch:\n\tGot: ${actualSpanNames}\n\tWant: ${expectedSpanNames}`
);

const expectedEventNames = [
'Starting stream',
'Transaction Creation Done',
'Acquiring session',
'Cache hit: has usable session',
'Acquired session',
'Using Session',
'Transaction Attempt Succeeded',
];
assert.deepStrictEqual(
actualEventNames,
expectedEventNames,
`Unexpected events:\n\tGot: ${actualEventNames}\n\tWant: ${expectedEventNames}`
);
const expectedEventNames = [
'Starting stream',
'Transaction Creation Done',
'Acquiring session',
'Cache hit: has usable session',
'Acquired session',
'Using Session',
'Transaction Attempt Succeeded',
];
assert.deepStrictEqual(
actualEventNames,
expectedEventNames,
`Unexpected events:\n\tGot: ${actualEventNames}\n\tWant: ${expectedEventNames}`
);
});
});

it('writeAtLeastOnce', done => {
Expand Down Expand Up @@ -1508,7 +1510,6 @@ SELECT 1p
'CloudSpanner.Snapshot.begin',
'CloudSpanner.Transaction.commit',
'CloudSpanner.Transaction.commit',
'CloudSpanner.Database.runTransactionAsync',
];
assert.deepStrictEqual(
actualSpanNames,
Expand Down Expand Up @@ -2125,7 +2126,6 @@ describe('Traces for ExecuteStream broken stream retries', () => {
'CloudSpanner.Snapshot.begin',
'CloudSpanner.Transaction.commit',
'CloudSpanner.Transaction.commit',
'CloudSpanner.Database.runTransactionAsync',
];
assert.deepStrictEqual(
actualSpanNames,
Expand Down Expand Up @@ -2207,7 +2207,6 @@ describe('Traces for ExecuteStream broken stream retries', () => {
const expectedSpanNames = [
'CloudSpanner.Database.batchCreateSessions',
'CloudSpanner.SessionPool.createSessions',
'CloudSpanner.Database.runTransactionAsync',
];
assert.deepStrictEqual(
actualSpanNames,
Expand Down
81 changes: 37 additions & 44 deletions src/database.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3385,53 +3385,46 @@ class Database extends common.GrpcServiceObject {

let sessionId = '';
const getSession = this.pool_.getSession.bind(this.pool_);
const span = getActiveOrNoopSpan();
// Loop to retry 'Session not found' errors.
// (and yes, we like while (true) more than for (;;) here)
// eslint-disable-next-line no-constant-condition
while (true) {
try {
const [session, transaction] = await promisify(getSession)();
transaction.requestOptions = Object.assign(
transaction.requestOptions || {},
options.requestOptions
);
if (options.optimisticLock) {
transaction.useOptimisticLock();
}
if (options.excludeTxnFromChangeStreams) {
transaction.excludeTxnFromChangeStreams();
}
sessionId = session?.id;
span.addEvent('Using Session', {'session.id': sessionId});
const runner = new AsyncTransactionRunner<T>(
session,
transaction,
runFn,
options
);

return startTrace(
'Database.runTransactionAsync',
this._traceConfig,
async span => {
// Loop to retry 'Session not found' errors.
// (and yes, we like while (true) more than for (;;) here)
// eslint-disable-next-line no-constant-condition
while (true) {
try {
const [session, transaction] = await promisify(getSession)();
transaction.requestOptions = Object.assign(
transaction.requestOptions || {},
options.requestOptions
);
if (options.optimisticLock) {
transaction.useOptimisticLock();
}
if (options.excludeTxnFromChangeStreams) {
transaction.excludeTxnFromChangeStreams();
}
sessionId = session?.id;
span.addEvent('Using Session', {'session.id': sessionId});
const runner = new AsyncTransactionRunner<T>(
session,
transaction,
runFn,
options
);

try {
const result = await runner.run();
span.end();
return result;
} finally {
this.pool_.release(session);
}
} catch (e) {
if (!isSessionNotFoundError(e as ServiceError)) {
setSpanErrorAndException(span, e as Error);
span.end();
throw e;
}
}
try {
return await runner.run();
} finally {
this.pool_.release(session);
}
} catch (e) {
if (!isSessionNotFoundError(e as ServiceError)) {
span.addEvent('No session available', {
'session.id': sessionId,
});
throw e;
}
}
);
}
}

/**
Expand Down

0 comments on commit 13bc9f5

Please sign in to comment.