unfinishedTransactions = forceCloseTransactionExecutor();
this.transactionExecutor = transactionExecutorServiceProvider.get();
setState(State.PATCHING);
setupDataSource();
setupDatabase();
for (Runnable unfinishedTransaction : unfinishedTransactions) {
transactionExecutor.submit(unfinishedTransaction);
}
// If an OperationCriticalTransaction fails open is set to false.
// See executeTransaction method below.
if (getState() == State.CLOSED) {
throw new DBInitException("Failed to set-up Database");
}
}
protected boolean attemptToCloseTransactionExecutor() {
if (transactionExecutor == null || transactionExecutor.isShutdown() || transactionExecutor.isTerminated()) {
return true;
}
transactionExecutor.shutdown();
try {
logger.info(locale.getString(PluginLang.DISABLED_WAITING_TRANSACTIONS));
Long waitMs = config.getOrDefault(TimeSettings.DB_TRANSACTION_FINISH_WAIT_DELAY, TimeUnit.SECONDS.toMillis(20L));
if (waitMs > TimeUnit.MINUTES.toMillis(5L)) {
logger.warn(TimeSettings.DB_TRANSACTION_FINISH_WAIT_DELAY.getPath() + " was set to over 5 minutes, using 5 min instead.");
waitMs = TimeUnit.MINUTES.toMillis(5L);
}
return transactionExecutor.awaitTermination(waitMs, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return true;
}
Patch[] patches() {
return new Patch[]{
new Version10Patch(),
new GeoInfoLastUsedPatch(),
new SessionAFKTimePatch(),
new KillsServerIDPatch(),
new WorldTimesSeverIDPatch(),
new WorldsServerIDPatch(),
new NicknameLastSeenPatch(),
new VersionTableRemovalPatch(),
new DiskUsagePatch(),
new WorldsOptimizationPatch(),
new KillsOptimizationPatch(),
new NicknamesOptimizationPatch(),
new TransferTableRemovalPatch(),
// new BadAFKThresholdValuePatch(),
new DeleteIPsPatch(),
new ExtensionShowInPlayersTablePatch(),
new ExtensionTableRowValueLengthPatch(),
new CommandUsageTableRemovalPatch(),
new BadNukkitRegisterValuePatch(),
new LinkedToSecurityTablePatch(),
new LinkUsersToPlayersSecurityTablePatch(),
new LitebansTableHeaderPatch(),
new UserInfoHostnamePatch(),
new ServerIsProxyPatch(),
new ServerTableRowPatch(),
new PlayerTableRowPatch(),
new ExtensionTableProviderValuesForPatch(),
new RemoveIncorrectTebexPackageDataPatch(),
new ExtensionTableProviderFormattersPatch(),
new ServerPlanVersionPatch(),
new RemoveDanglingUserDataPatch(),
new RemoveDanglingServerDataPatch(),
new GeoInfoOptimizationPatch(),
new PingOptimizationPatch(),
new UserInfoOptimizationPatch(),
new WorldTimesOptimizationPatch(),
new SessionsOptimizationPatch(),
new UserInfoHostnameAllowNullPatch(),
new RegisterDateMinimizationPatch(),
new UsersTableNameLengthPatch(),
new SessionJoinAddressPatch(),
new RemoveUsernameFromAccessLogPatch(),
new ComponentColumnToExtensionDataPatch(),
new BadJoinAddressDataCorrectionPatch(),
new AfterBadJoinAddressDataCorrectionPatch(),
new CorrectWrongCharacterEncodingPatch(logger, config),
new UpdateWebPermissionsPatch(),
new WebGroupDefaultGroupsPatch(),
new WebGroupAddMissingAdminGroupPatch(),
new LegacyPermissionLevelGroupsPatch(),
new SecurityTableGroupPatch()
};
}
/**
* Ensures connection functions correctly and all tables exist.
*
* Updates to latest schema.
*/
private void setupDatabase() {
executeTransaction(new OperationCriticalTransaction() {
@Override
protected void performOperations() {
logger.info(locale.getString(PluginLang.DB_SCHEMA_PATCH));
}
});
executeTransaction(new CreateTablesTransaction());
for (Patch patch : patches()) {
executeTransaction(patch);
}
executeTransaction(new OperationCriticalTransaction() {
@Override
protected void performOperations() {
logger.info(locale.getString(PluginLang.DB_APPLIED_PATCHES));
if (getState() == State.PATCHING) setState(State.OPEN);
}
});
registerIndexCreationTask();
}
private void registerIndexCreationTask() {
try {
runnableFactory.create(new PluginRunnable() {
@Override
public void run() {
if (getState() == State.CLOSED || getState() == State.CLOSING) {
cancel();
return;
}
try {
executeTransaction(new CreateIndexTransaction());
} catch (DBOpException e) {
errorLogger.warn(e);
}
}
}).runTaskLaterAsynchronously(TimeAmount.toTicks(1, TimeUnit.MINUTES));
} catch (Exception ignore) {
// Task failed to register because plugin is being disabled
}
}
/**
* Set up the source for connections.
*
* @throws DBInitException If the DataSource fails to be initialized.
*/
public abstract void setupDataSource();
protected List forceCloseTransactionExecutor() {
if (transactionExecutor == null || transactionExecutor.isShutdown() || transactionExecutor.isTerminated()) {
return Collections.emptyList();
}
try {
List unfinished = transactionExecutor.shutdownNow();
int unfinishedCount = unfinished.size();
if (unfinishedCount > 0) {
logger.warn(unfinishedCount + " unfinished database transactions were not executed.");
}
return unfinished;
} finally {
logger.info(locale.getString(PluginLang.DISABLED_WAITING_TRANSACTIONS_COMPLETE));
}
}
@Override
public void close() {
// SQLiteDB Overrides this, so any additions to this should also be reflected there.
if (getState() == State.OPEN) setState(State.CLOSING);
if (attemptToCloseTransactionExecutor()) {
logger.info(locale.getString(PluginLang.DISABLED_WAITING_TRANSACTIONS_COMPLETE));
} else {
forceCloseTransactionExecutor();
}
unloadDriverClassloader();
setState(State.CLOSED);
}
public abstract Connection getConnection() throws SQLException;
public abstract void returnToPool(Connection connection);
@Override
public T query(Query query) {
return accessLock.performDatabaseOperation(() -> query.executeQuery(this));
}
public T queryWithinTransaction(Query query, Transaction transaction) {
return accessLock.performDatabaseOperation(() -> query.executeQuery(this), transaction);
}
protected void unloadDriverClassloader() {
// Unloading class loader using close() causes issues when reloading.
// It is better to leak this memory than crash the plugin on reload.
driverClassLoader = null;
}
@Override
public CompletableFuture> executeTransaction(Transaction transaction) {
if (getState() == State.CLOSED) {
throw new DBClosedException("Transaction tried to execute although database is closed.");
}
StackTraceElement[] origin = Thread.currentThread().getStackTrace();
if (determineIfShouldDropUnimportantTransactions(transactionQueueSize.incrementAndGet())
&& transaction instanceof ThrowawayTransaction) {
// Drop throwaway transaction immediately.
return CompletableFuture.completedFuture(null);
}
return CompletableFuture.supplyAsync(() -> {
try {
TRANSACTION_ORIGIN.set(origin);
if (getState() == State.CLOSED) return CompletableFuture.completedFuture(null);
accessLock.performDatabaseOperation(() -> {
if (!ranIntoFatalError.get()) {transaction.executeTransaction(this);}
}, transaction);
return CompletableFuture.completedFuture(null);
} finally {
transactionQueueSize.decrementAndGet();
TRANSACTION_ORIGIN.remove();
}
}, getTransactionExecutor()).exceptionally(errorHandler(transaction, origin));
}
private boolean determineIfShouldDropUnimportantTransactions(int queueSize) {
if (getState() == State.CLOSING) {
return true;
}
boolean dropTransactions = dropUnimportantTransactions.get();
if (queueSize >= 500 && !dropTransactions) {
logger.warn("Database queue size: " + queueSize + ", dropping some unimportant transactions. If this keeps happening disable some extensions or optimize MySQL.");
dropUnimportantTransactions.set(true);
return true;
} else if (queueSize < 50 && dropTransactions) {
dropUnimportantTransactions.set(false);
return false;
}
return dropTransactions;
}
private Function> errorHandler(Transaction transaction, StackTraceElement[] origin) {
return throwable -> {
if (throwable == null) {
return CompletableFuture.completedFuture(null);
}
if (throwable.getCause() instanceof FatalDBException) {
ranIntoFatalError.set(true);
logger.error("Database failed to open, " + transaction.getClass().getName() + " failed to be executed.");
FatalDBException actual = (FatalDBException) throwable.getCause();
Optional whatToDo = actual.getContext().flatMap(ErrorContext::getWhatToDo);
whatToDo.ifPresentOrElse(
message -> logger.error("What to do: " + message),
() -> logger.error("Error msg: " + actual.getMessage())
);
setState(State.CLOSED);
}
ThrowableUtils.appendEntryPointToCause(throwable, origin);
ErrorContext errorContext = ErrorContext.builder()
.related("Transaction: " + transaction.getClass())
.related("DB State: " + getState() + " - fatal: " + ranIntoFatalError.get())
.build();
if (getState() == State.CLOSED) {
errorLogger.critical(throwable, errorContext);
} else {
errorLogger.error(throwable, errorContext);
}
return CompletableFuture.completedFuture(null);
};
}
private ExecutorService getTransactionExecutor() {
if (transactionExecutor == null) {
transactionExecutor = transactionExecutorServiceProvider.get();
}
return transactionExecutor;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
SQLDB sqldb = (SQLDB) o;
return getType() == sqldb.getType();
}
@Override
public int hashCode() {
return Objects.hash(getType().getName());
}
public Supplier getServerUUIDSupplier() {
return serverUUIDSupplier;
}
public void setTransactionExecutorServiceProvider(Supplier transactionExecutorServiceProvider) {
this.transactionExecutorServiceProvider = transactionExecutorServiceProvider;
}
public RunnableFactory getRunnableFactory() {
return runnableFactory;
}
public PluginLogger getLogger() {
return logger;
}
public Locale getLocale() {
return locale;
}
public boolean shouldDropUnimportantTransactions() {
return dropUnimportantTransactions.get();
}
public int getTransactionQueueSize() {
return transactionQueueSize.get();
}
}