CORDA-1812 - fix Postgres db bloat issue (#3614) (#3857)

node-api/src/main/kotlin/net/corda/nodeapi/internal/persistence/CordaPersistence.kt

@@ -84,6 +84,7 @@ class CordaPersistence(
             check(!connection.metaData.isReadOnly) { "Database should not be readonly." }
 
             checkCorrectAttachmentsContractsTableName(connection)
+            checkCorrectCheckpointTypeOnPostgres(connection)
         }
     }
 
@@ -250,6 +251,18 @@ fun <T : Any> rx.Observable<T>.wrapWithDatabaseTransaction(db: CordaPersistence?
 
 class IncompatibleAttachmentsContractsTableName(override val message: String?, override val cause: Throwable? = null) : Exception()
 
+/** Check if any nested cause is of [SQLException] type. */
+private fun Throwable.hasSQLExceptionCause(): Boolean =
+        when (cause) {
+            null -> false
+            is SQLException -> true
+            else -> cause?.hasSQLExceptionCause() ?: false
+        }
+
+class CouldNotCreateDataSourceException(override val message: String?, override val cause: Throwable? = null) : Exception()
+
+class DatabaseIncompatibleException(override val message: String?, override val cause: Throwable? = null) : Exception()
+
 private fun checkCorrectAttachmentsContractsTableName(connection: Connection) {
     val correctName = "NODE_ATTACHMENTS_CONTRACTS"
     val incorrectV30Name = "NODE_ATTACHMENTS_CONTRACT_CLASS_NAME"
@@ -258,7 +271,22 @@ private fun checkCorrectAttachmentsContractsTableName(connection: Connection) {
     fun warning(incorrectName: String, version: String) = "The database contains the older table name $incorrectName instead of $correctName, see upgrade notes to migrate from Corda database version $version https://docs.corda.net/head/upgrade-notes.html."
 
     if (!connection.metaData.getTables(null, null, correctName, null).next()) {
-        if (connection.metaData.getTables(null, null, incorrectV30Name, null).next()) { throw IncompatibleAttachmentsContractsTableName(warning(incorrectV30Name, "3.0")) }
+        if (connection.metaData.getTables(null, null, incorrectV30Name, null).next()) { throw DatabaseIncompatibleException(warning(incorrectV30Name, "3.0")) }
-        if (connection.metaData.getTables(null, null, incorrectV31Name, null).next()) { throw IncompatibleAttachmentsContractsTableName(warning(incorrectV31Name, "3.1")) }
+        if (connection.metaData.getTables(null, null, incorrectV31Name, null).next()) { throw DatabaseIncompatibleException(warning(incorrectV31Name, "3.1")) }
+    }
+}
+
+private fun checkCorrectCheckpointTypeOnPostgres(connection: Connection) {
+    val metaData = connection.metaData
+    if (metaData.getDatabaseProductName() != "PostgreSQL") {
+        return
+    }
+
+    val result = metaData.getColumns(null, null, "node_checkpoints", "checkpoint_value")
+    if (result.next()) {
+        val type = result.getString("TYPE_NAME")
+        if (type != "bytea") {
+            throw DatabaseIncompatibleException("The type of the 'checkpoint_value' table must be 'bytea', but 'oid' was found. See upgrade notes to migrate from Corda database version 3.1 https://docs.corda.net/head/upgrade-notes.html.")
+        }
     }
 }

node/src/integration-test/kotlin/net/corda/node/persistence/FailNodeOnNotMigratedAttachmentContractsTableNameTests.kt

@@ -49,7 +49,7 @@ class FailNodeOnNotMigratedAttachmentContractsTableNameTests {
                 it.createStatement().execute("ALTER TABLE $tableNameFromMapping RENAME TO $tableNameInDB")
                 it.commit()
             }
-            assertFailsWith(net.corda.nodeapi.internal.persistence.IncompatibleAttachmentsContractsTableName::class) {
+            assertFailsWith(net.corda.nodeapi.internal.persistence.DatabaseIncompatibleException::class) {
                 val nodeHandle = startNode(providedName = nodeName, rpcUsers = listOf(user)).getOrThrow()
                 nodeHandle.stop()
             }

node/src/main/kotlin/net/corda/node/internal/NodeStartup.kt

@@ -17,7 +17,7 @@ import net.corda.node.shell.InteractiveShell
 import net.corda.node.utilities.registration.HTTPNetworkRegistrationService
 import net.corda.node.utilities.registration.NetworkRegistrationHelper
 import net.corda.nodeapi.internal.addShutdownHook
-import net.corda.nodeapi.internal.persistence.IncompatibleAttachmentsContractsTableName
+import net.corda.nodeapi.internal.persistence.DatabaseIncompatibleException
 import org.fusesource.jansi.Ansi
 import org.fusesource.jansi.AnsiConsole
 import org.slf4j.bridge.SLF4JBridgeHandler
@@ -114,7 +114,7 @@ open class NodeStartup(val args: Array<String>) {
         try {
             cmdlineOptions.baseDirectory.createDirectories()
             startNode(conf, versionInfo, startTime, cmdlineOptions)
-        } catch (e: IncompatibleAttachmentsContractsTableName) {
+        } catch (e: DatabaseIncompatibleException) {
             e.message?.let { Node.printWarning(it) }
             logger.error(e.message)
             return false

node/src/main/kotlin/net/corda/node/services/persistence/DBCheckpointStorage.kt

@@ -9,7 +9,7 @@ import java.io.Serializable
 import javax.persistence.Column
 import javax.persistence.Entity
 import javax.persistence.Id
-import javax.persistence.Lob
+import org.hibernate.annotations.Type
 
 /**
  * Simple checkpoint key value storage in DB.
@@ -23,8 +23,8 @@ class DBCheckpointStorage : CheckpointStorage {
             @Column(name = "checkpoint_id", length = 64, nullable = false)
             var checkpointId: String = "",
 
-            @Lob
+            @Column(name = "checkpoint_value")
-            @Column(name = "checkpoint_value", nullable = false)
+            @Type(type="org.hibernate.type.ImageType")
             var checkpoint: ByteArray = ByteArray(0)
     ) : Serializable