Visitar URL original
fix(#11068, #11067, #11059): a close for reopen refuses new transacti… · ArcadeData/arcadedb@fcf9c73 · GitHub
Skip to content

Commit fcf9c73

Browse files
lvcalvca
committed
fix(#11068, #11067, #11059): a close for reopen refuses new transactions, nested rows keep their own columns, one JSON soft-limit budget per request, the opening chunk of a load runs bound
1 parent 9a0c131 commit fcf9c73

7 files changed

Lines changed: 246 additions & 10 deletions

File tree

‎engine/src/main/java/com/arcadedb/database/LocalDatabase.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -837,6 +837,14 @@ public void begin() {
837837
public void begin(final TRANSACTION_ISOLATION_LEVEL isolationLevel) {
838838
executeInReadLock(() -> {
839839
checkDatabaseIsOpen();
840+
// A CLOSE FOR REOPEN HAS STARTED (AN IDLE CLOSE, issue #11068; AN HA REINSTALL): THE INSTANCE IS STILL OPEN WHILE IT DRAINS
841+
// ITS ASYNCHRONOUS WORK, BUT ITS TEARDOWN WOULD ROLL BACK A TRANSACTION BEGUN NOW. REFUSED AS RETRYABLE: THE RETRY FINDS THE
842+
// DATABASE REOPENED. NOT TO THE ASYNCHRONOUS WORKERS, WHOSE BATCHES ARE WHAT THE DRAIN WAITS FOR
843+
if (closedForReopen) {
844+
final DatabaseAsyncExecutorImpl executor = async;
845+
if (executor == null || !executor.isCurrentThreadOneOfMyWorkers())
846+
throw new DatabaseIsReopeningException(name);
847+
}
840848

841849
// FORCE THE RESET OF TL
842850
final DatabaseContext.DatabaseContextTL current = DatabaseContext.INSTANCE.getContext(databasePath);

‎engine/src/test/java/com/arcadedb/database/Issue11068BeginDuringCloseForReopenTest.java‎

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,94 @@
1+
/*
2+
* Copyright © 2021-present Arcade Data Ltd (info@arcadedata.com)
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*
16+
* SPDX-FileCopyrightText: 2021-present Arcade Data Ltd (info@arcadedata.com)
17+
* SPDX-License-Identifier: Apache-2.0
18+
*/
19+
package com.arcadedb.database;
20+
21+
import com.arcadedb.exception.DatabaseIsReopeningException;
22+
import org.junit.jupiter.api.AfterEach;
23+
import org.junit.jupiter.api.Test;
24+
import org.junit.jupiter.api.io.TempDir;
25+
26+
import java.nio.file.Path;
27+
import java.util.concurrent.CompletableFuture;
28+
import java.util.concurrent.CountDownLatch;
29+
import java.util.concurrent.TimeUnit;
30+
import java.util.concurrent.atomic.AtomicReference;
31+
32+
import static org.assertj.core.api.Assertions.assertThat;
33+
34+
/**
35+
* Issue #11068: a close FOR REOPEN (a server closing an idle database, an HA reinstall) keeps the instance open while it drains
36+
* its asynchronous work, before the write-locked teardown that rolls back every transaction still attached to it. A handle
37+
* resolved before the close could begin a transaction in that window and lose it to the teardown. Such a {@code begin()} is now
38+
* refused with the retryable {@link DatabaseIsReopeningException}, so the caller retries on the reopened database.
39+
*
40+
* @author Luca Garulli (l.garulli@arcadedata.com)
41+
*/
42+
class Issue11068BeginDuringCloseForReopenTest {
43+
44+
@TempDir
45+
private Path tempDir;
46+
private LocalDatabase database;
47+
48+
@AfterEach
49+
void tearDown() {
50+
LocalDatabase.TEST_CLOSE_HOOK = null;
51+
if (database != null && database.isOpen())
52+
database.close();
53+
}
54+
55+
@Test
56+
void aTransactionBegunWhileTheInstanceClosesForReopenIsRefusedAsRetryable() throws Exception {
57+
database = (LocalDatabase) new DatabaseFactory(tempDir.resolve("reopening").toString()).create();
58+
database.transaction(() -> database.getSchema().createDocumentType("Doc"));
59+
60+
final CountDownLatch insideClose = new CountDownLatch(1);
61+
final CountDownLatch releaseClose = new CountDownLatch(1);
62+
LocalDatabase.TEST_CLOSE_HOOK = () -> {
63+
insideClose.countDown();
64+
try {
65+
releaseClose.await(30, TimeUnit.SECONDS);
66+
} catch (final InterruptedException e) {
67+
Thread.currentThread().interrupt();
68+
}
69+
};
70+
71+
final Thread closer = new Thread(database::closeForReopen, "closer-for-reopen");
72+
closer.start();
73+
try {
74+
assertThat(insideClose.await(30, TimeUnit.SECONDS)).isTrue();
75+
assertThat(database.isOpen()).as("still open: the teardown has not run yet").isTrue();
76+
77+
final AtomicReference<Throwable> refused = new AtomicReference<>();
78+
CompletableFuture.runAsync(() -> {
79+
try {
80+
database.begin();
81+
database.rollback();
82+
} catch (final Throwable e) {
83+
refused.set(e);
84+
}
85+
}).get(30, TimeUnit.SECONDS);
86+
87+
assertThat(refused.get()).isInstanceOf(DatabaseIsReopeningException.class);
88+
} finally {
89+
releaseClose.countDown();
90+
closer.join(30_000);
91+
}
92+
assertThat(database.isOpen()).isFalse();
93+
}
94+
}

‎grpcw/src/main/java/com/arcadedb/server/grpc/ArcadeDbGrpcService.java‎

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -5615,8 +5615,9 @@ private GrpcValue toGrpcValue(Object o, ProjectionConfig pc) {
56155615

56165616
byte[] jsonBytes = json.toString().getBytes(StandardCharsets.UTF_8);
56175617

5618-
// Soft limit handling
5619-
if (pc.softLimitBytes > 0 && jsonBytes.length > pc.softLimitBytes) {
5618+
// Soft limit handling: against what is LEFT of the request's budget, as the MAP encoding counts it. Each record checked
5619+
// against the whole budget let a list of them, each under the limit, add up to many times it (issue #11048)
5620+
if (pc.softLimitBytes > 0 && pc.used.get() + jsonBytes.length > pc.softLimitBytes) {
56205621
pc.truncated = true;
56215622
if (debug)
56225623
LogManager.instance().log(this, Level.FINE, """
@@ -5718,11 +5719,20 @@ private GrpcValue toGrpcValue(Object o, ProjectionConfig pc) {
57185719

57195720
// Support ArcadeDB Result/ResultInternal as structural MAP
57205721
if (o instanceof Result res) {
5721-
// A result that wraps a record (a sub-query row, a nested match) is that record, @rid and @type included, as HTTP sends it
5722-
if (res.isElement())
5723-
return toGrpcValue(res.toElement(), pc);
5724-
57255722
GrpcMap.Builder mb = GrpcMap.newBuilder();
5723+
if (res.isElement()) {
5724+
final Document element = res.toElement();
5725+
// A result that is just a record (a sub-query row) is that record, encoded as the request asks for records
5726+
if (res.getPropertyNames().equals(element.getPropertyNames()))
5727+
return toGrpcValue(element, pc);
5728+
// A row with columns of its own besides the record (SELECT *, name AS alias; a Cypher RETURN n row): its columns, with the
5729+
// record's @rid and @type, as the HTTP serializer sends it
5730+
if (element.getIdentity() != null && element.getIdentity().isValid())
5731+
mb.putEntries(Property.RID_PROPERTY, GrpcValue.newBuilder()
5732+
.setLinkValue(GrpcLink.newBuilder().setRid(element.getIdentity().toString()).build()).setLogicalType("rid").build());
5733+
if (element.getType() != null)
5734+
mb.putEntries(Property.TYPE_PROPERTY, GrpcValue.newBuilder().setStringValue(element.getTypeName()).build());
5735+
}
57265736
try {
57275737
for (String k : res.getPropertyNames()) {
57285738
mb.putEntries(k, toGrpcValue(res.getProperty(k), pc));

‎grpcw/src/main/java/com/arcadedb/server/grpc/StreamingLoadEntry.java‎

Lines changed: 19 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,8 @@
3535
* {@code PER_STREAM} transaction mode holds one write transaction from its first chunk to its last, between chunks
3636
* too, and that is what an administrator has to be able to see and stop.
3737
* <p>
38-
* <b>How a termination reaches the load.</b> Every chunk runs with the entry bound on its thread, so the rows' checks and
38+
* <b>How a termination reaches the load.</b> Every chunk runs with the entry bound on its thread - the first one from the
39+
* moment it {@link #open opens} the entry, since there is none to bind when it enters - so the rows' checks and
3940
* the commits the chunk attempts obey it as any statement's do. A load that is idle between two chunks is reached by the
4041
* termination listener given to {@link #open}, which must never act under a running chunk: it closes the call, and the
4142
* handler retracts what the stream's mode can still retract on the stream's own thread, where its transaction lives,
@@ -63,6 +64,8 @@ final class StreamingLoadEntry {
6364
/** Held by a chunk while it runs, so the termination listener of an idle stream never acts under a running chunk. */
6465
private final ReentrantLock working = new ReentrantLock();
6566
private volatile RunningQuery entry;
67+
/** What {@link #open} bound for the chunk that opened the entry, inside {@link #enter()}: read and written under {@link #working}. */
68+
private RunningQuery.Binding openingBinding;
6669

6770
/**
6871
* Must be built on the call's gRPC Context - inside the RPC handler, as every caller does - since that is where
@@ -94,6 +97,10 @@ void open(final String database, final String user, final String sessionId, fina
9497
opened.setStatement("grpc", text);
9598
entry = opened;
9699
opened.onTerminate(onTerminate);
100+
// The chunk that opens the entry is bound for the rest of its work like every later chunk: it entered before there was an
101+
// entry to bind, so its first rows and the commit they reach would not see a termination. Closed by that chunk's binding
102+
if (working.isHeldByCurrentThread())
103+
openingBinding = opened.bind();
97104
}
98105

99106
/**
@@ -106,7 +113,17 @@ Bound enter() {
106113
working.lock();
107114
final RunningQuery current = entry;
108115
if (current == null)
109-
return working::unlock;
116+
// The chunk that may open the entry: what open() bound for it is closed with it
117+
return () -> {
118+
try {
119+
final RunningQuery.Binding opening = openingBinding;
120+
openingBinding = null;
121+
if (opening != null)
122+
opening.close();
123+
} finally {
124+
working.unlock();
125+
}
126+
};
110127
final RunningQuery.Binding binding = current.bind();
111128
return () -> {
112129
try {

‎grpcw/src/test/java/com/arcadedb/server/grpc/Issue11059StreamingLoadFirstChunkBindingTest.java‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
/*
2+
* Copyright © 2021-present Arcade Data Ltd (info@arcadedata.com)
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*
16+
* SPDX-FileCopyrightText: 2021-present Arcade Data Ltd (info@arcadedata.com)
17+
* SPDX-License-Identifier: Apache-2.0
18+
*/
19+
package com.arcadedb.server.grpc;
20+
21+
import com.arcadedb.query.RunningQuery;
22+
import com.arcadedb.query.RunningQueryRegistry;
23+
import org.junit.jupiter.api.Test;
24+
25+
import static org.assertj.core.api.Assertions.assertThat;
26+
27+
/**
28+
* Issue #11059, follow-up: the chunk that opens a load's running-query entry enters before there is an entry to bind, so its
29+
* rows and the commit they reach ran unbound, and a termination arriving meanwhile was seen only by the next chunk. The entry is
30+
* now bound on the opening chunk's thread from the moment it is opened, and unbound with that chunk.
31+
*
32+
* @author Luca Garulli (l.garulli@arcadedata.com)
33+
*/
34+
class Issue11059StreamingLoadFirstChunkBindingTest {
35+
36+
@Test
37+
void theChunkThatOpensTheEntryRunsBound() {
38+
final RunningQueryRegistry registry = new RunningQueryRegistry();
39+
final StreamingLoadEntry entry = new StreamingLoadEntry(registry, "TimeSeriesWriteStream");
40+
41+
final StreamingLoadEntry.Bound first = entry.enter();
42+
try {
43+
assertThat(RunningQuery.current()).as("nothing to bind before the entry is opened").isNull();
44+
entry.open("db", "root", null, () -> "TimeSeriesWriteStream", () -> {
45+
});
46+
assertThat(RunningQuery.current()).as("the rest of the opening chunk runs bound").isNotNull();
47+
assertThat(RunningQuery.current().getDatabase()).isEqualTo("db");
48+
} finally {
49+
first.close();
50+
}
51+
assertThat(RunningQuery.current()).as("unbound with the chunk").isNull();
52+
53+
// A later chunk binds it as before
54+
final StreamingLoadEntry.Bound second = entry.enter();
55+
try {
56+
assertThat(RunningQuery.current()).isNotNull();
57+
} finally {
58+
second.close();
59+
}
60+
assertThat(RunningQuery.current()).isNull();
61+
entry.close();
62+
}
63+
64+
@Test
65+
void anEntryOpenedOutsideAChunkIsNotBoundOnTheOpeningThread() {
66+
// The bidirectional insert opens its entry from a message that took no working lock: nothing would close such a binding
67+
final StreamingLoadEntry entry = new StreamingLoadEntry(new RunningQueryRegistry(), "InsertBidirectional");
68+
entry.open("db", "root", null, () -> "InsertBidirectional", () -> {
69+
});
70+
assertThat(RunningQuery.current()).isNull();
71+
entry.close();
72+
}
73+
}

‎grpcw/src/test/java/com/arcadedb/server/grpc/Issue11067GrpcTraversalProjectionIT.java‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -152,6 +152,39 @@ void aNestedResultRowIsTheRecordItWraps() {
152152
assertThat(rids(value)).containsExactly(ridB);
153153
}
154154

155+
@Test
156+
void aNestedRowWithColumnsOfItsOwnKeepsThem() {
157+
// SELECT *, name AS alias keeps the record AND publishes a column of its own: collapsing the row to the record lost it
158+
final GrpcValue value = project("(SELECT *, name AS alias FROM " + VERTEX + " WHERE name = 'b')", "a", null);
159+
assertThat(value.getListValue().getValuesCount()).isEqualTo(1);
160+
final GrpcMap row = value.getListValue().getValues(0).getMapValue();
161+
assertThat(row.getEntriesMap().get("alias").getStringValue()).isEqualTo("b");
162+
assertThat(row.getEntriesMap().get("@rid").getLinkValue().getRid()).isEqualTo(ridB);
163+
assertThat(row.getEntriesMap().get("@type").getStringValue()).isEqualTo(VERTEX);
164+
}
165+
166+
@Test
167+
void theSoftLimitIsOneBudgetForEveryRecordOfTheList() {
168+
final Database database = getServerDatabase(0, getDatabaseName());
169+
database.transaction(() -> {
170+
final MutableVertex hub = database.newVertex(VERTEX).set("name", "hub").save();
171+
hub.newEdge(EDGE, database.newVertex(VERTEX).set("name", "x").save());
172+
hub.newEdge(EDGE, database.newVertex(VERTEX).set("name", "y").save());
173+
});
174+
175+
final GrpcValue unbounded = project("out('" + EDGE + "')", "hub", projection(ProjectionEncoding.PROJECTION_AS_JSON, 0));
176+
assertThat(unbounded.getListValue().getValuesCount()).isEqualTo(2);
177+
final int first = unbounded.getListValue().getValues(0).getBytesValue().size();
178+
final int second = unbounded.getListValue().getValues(1).getBytesValue().size();
179+
180+
// Room for the first record, not for both: each one fits the limit alone, the two do not fit it together
181+
final GrpcValue bounded = project("out('" + EDGE + "')", "hub",
182+
projection(ProjectionEncoding.PROJECTION_AS_JSON, first + second - 1));
183+
assertThat(bounded.getListValue().getValues(0).getKindCase()).isEqualTo(GrpcValue.KindCase.BYTES_VALUE);
184+
assertThat(bounded.getListValue().getValues(1).getKindCase()).as("over the budget: its link")
185+
.isEqualTo(GrpcValue.KindCase.LINK_VALUE);
186+
}
187+
155188
@Test
156189
void theSoftLimitBoundsARecordInsideAList() {
157190
// A record the budget cannot hold goes out as its link (the JSON encoding's fallback): the budget sees a record nested

‎server/src/main/java/com/arcadedb/server/http/HttpSessionManager.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -267,7 +267,7 @@ public int getActiveSessions() {
267267
public boolean hasSessionOnDatabase(final String databaseName) {
268268
return executeInReadLock(() -> {
269269
for (final HttpSession session : sessions.values())
270-
if (databaseName.equals(session.transaction.getDatabase().getName()))
270+
if (session.transaction != null && databaseName.equals(session.transaction.getDatabase().getName()))
271271
return true;
272272
return false;
273273
});
@@ -278,7 +278,8 @@ public Set<String> getSessionDatabaseNames() {
278278
return executeInReadLock(() -> {
279279
final Set<String> names = new HashSet<>();
280280
for (final HttpSession session : sessions.values())
281-
names.add(session.transaction.getDatabase().getName());
281+
if (session.transaction != null)
282+
names.add(session.transaction.getDatabase().getName());
282283
return names;
283284
});
284285
}

0 commit comments

Comments
 (0)