async binaries lo_unlink

* Added new binaries_lo_unlink_queue table to temporary store large
object OIDs.
* ExecutorService in BinaryDaoJdbc is used to asynchronously delete rows
from binaries_lo_unlink_queue table. Table uses before delete trigger to
call lo_unlink.

Author: hhund

dsf-fhir/dsf-fhir-server/src/main/java/dev/dsf/fhir/dao/BinaryDao.java

@@ -15,4 +15,8 @@ public interface BinaryDao extends ResourceDao<Binary>
 
 	Optional<Binary> readVersion(UUID uuid, long version, RangeRequest rangeRequest)
 			throws SQLException, ResourceDeletedException;
+	
+	void startLargeObjectUnlink();
+	
+	void stopLargeObjectUnlinker();
 }

dsf-fhir/dsf-fhir-server/src/main/java/dev/dsf/fhir/dao/jdbc/BinaryDaoJdbc.java

@@ -9,6 +9,9 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 import javax.sql.DataSource;
 
@@ -34,6 +37,8 @@ public class BinaryDaoJdbc extends AbstractResourceDaoJdbc<Binary> implements Bi
 
 	private final String selectUpdateUser;
 
+	private final ExecutorService loUnlinker;
+
 	public BinaryDaoJdbc(DataSource dataSource, DataSource permanentDeleteDataSource, FhirContext fhirContext,
 			String selectUpdateUser)
 	{
@@ -44,6 +49,16 @@ public BinaryDaoJdbc(DataSource dataSource, DataSource permanentDeleteDataSource
 				List.of());
 
 		this.selectUpdateUser = selectUpdateUser;
+
+		loUnlinker = Executors.newFixedThreadPool(1, r -> new Thread(r, "binaries-large-object-unlinker"));
+	}
+
+	@Override
+	public void afterPropertiesSet() throws Exception
+	{
+		super.afterPropertiesSet();
+
+		startLargeObjectUnlink();
 	}
 
 	@Override
@@ -199,4 +214,49 @@ public Optional<Binary> readVersion(UUID uuid, long version, RangeRequest rangeR
 
 		return binary;
 	}
+
+	@Override
+	public void startLargeObjectUnlink()
+	{
+		loUnlinker.submit(this::doLargeObjectUnlink);
+	}
+
+	private void doLargeObjectUnlink()
+	{
+		logger.debug("Deleting entries from binaries_lo_unlink_queue");
+
+		try (Connection connection = getPermanentDeleteDataSource().getConnection();
+				PreparedStatement statement = connection.prepareStatement("DELETE FROM binaries_lo_unlink_queue"))
+		{
+			statement.execute();
+		}
+		catch (SQLException e)
+		{
+			logger.debug("Unable to delete entries from binaries_lo_unlink_queue table", e);
+			logger.error("Unable to delete entries from binaries_lo_unlink_queue table: {} - {}", e.getClass().getName(),
+					e.getMessage());
+		}
+	}
+
+	@Override
+	public void stopLargeObjectUnlinker()
+	{
+		startLargeObjectUnlink();
+
+		logger.debug("Shutting down binaries-large-object-unlinker executor ...");
+
+		loUnlinker.shutdown();
+
+		try
+		{
+			if (!loUnlinker.awaitTermination(60, TimeUnit.SECONDS))
+			{
+				loUnlinker.shutdownNow();
+			}
+		}
+		catch (InterruptedException ex)
+		{
+			loUnlinker.shutdownNow();
+		}
+	}
 }

dsf-fhir/dsf-fhir-server/src/main/java/dev/dsf/fhir/spring/config/DaoConfig.java

@@ -121,7 +121,7 @@ public ActivityDefinitionDao activityDefinitionDao()
 		return new ActivityDefinitionDaoJdbc(dataSource(), permanentDeleteDataSource(), fhirConfig.fhirContext());
 	}
 
-	@Bean
+	@Bean(destroyMethod = "stopLargeObjectUnlinker")
 	public BinaryDao binaryDao()
 	{
 		return new BinaryDaoJdbc(dataSource(), permanentDeleteDataSource(), fhirConfig.fhirContext(),

dsf-fhir/dsf-fhir-server/src/main/java/dev/dsf/fhir/webservice/impl/BinaryServiceImpl.java

@@ -167,18 +167,19 @@ protected MediaType getMediaTypeForRead(UriInfo uri, HttpHeaders headers)
 		return parameterConverter.getMediaTypeIfSupported(uri, headers).orElseGet(() -> getMediaType(headers));
 	}
 
-	// @Override
-	// protected MediaType getMediaTypeForRead(UriInfo uri, HttpHeaders headers)
-	// {
-	// if (uri.getQueryParameters().containsKey(Constants.PARAM_FORMAT))
-	// return super.getMediaTypeForRead(uri, headers);
-	// else
-	// return getMediaType(headers);
-	// }
-	//
 	private MediaType getMediaType(HttpHeaders headers)
 	{
 		List<MediaType> types = headers.getAcceptableMediaTypes();
 		return types == null ? null : types.get(0);
 	}
+
+	@Override
+	public Response deletePermanently(String deletePath, String id, UriInfo uri, HttpHeaders headers)
+	{
+		Response response = super.deletePermanently(deletePath, id, uri, headers);
+
+		dao.startLargeObjectUnlink();
+
+		return response;
+	}
 }

dsf-fhir/dsf-fhir-server/src/main/resources/fhir/db/db.binaries_lo_unlink_queue.changelog-2.0.0.xml

@@ -0,0 +1,33 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog
+                      http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-3.6.xsd">
+
+	<changeSet author="hhund" id="db.binaries_lo_unlink_queue.changelog-2.0.0">
+		<createTable tableName="binaries_lo_unlink_queue">
+			<column name="binary_oid" type="oid">
+				<constraints nullable="false" />
+			</column>
+			<column name="queued_at" type="TIMESTAMP" defaultValue="now()" />
+		</createTable>
+
+		<addPrimaryKey tableName="binaries_lo_unlink_queue" columnNames="binary_oid" />
+
+		<sql dbms="postgresql">
+			ALTER TABLE binaries_lo_unlink_queue OWNER TO ${db.liquibase_user};
+			GRANT ALL ON TABLE binaries_lo_unlink_queue TO ${db.liquibase_user};
+			GRANT SELECT, INSERT, DELETE ON TABLE binaries_lo_unlink_queue TO ${db.server_permanent_delete_users_group};
+			CREATE INDEX binary_oid_index ON binaries_lo_unlink_queue USING btree (binary_oid);
+		</sql>
+	</changeSet>
+	
+	<changeSet author="hhund" id="db.binaries_lo_unlink_queue.changelog-2.0.0.on_binaries_lo_unlink_queue_delete_function" runOnChange="true">
+		<sqlFile dbms="postgresql" relativeToChangelogFile="true" path="trigger_functions/on_binaries_lo_unlink_queue_delete.sql" splitStatements="false" />
+	</changeSet>
+	
+	<changeSet author="hhund" id="db.binaries_lo_unlink_queue.changelog-2.0.0.on_delete_trigger">
+		<sql dbms="postgresql">
+			CREATE TRIGGER binaries_lo_unlink_queue_delete BEFORE DELETE ON binaries_lo_unlink_queue FOR EACH ROW EXECUTE PROCEDURE on_binaries_lo_unlink_queue_delete();
+		</sql>
+	</changeSet>
+</databaseChangeLog>

dsf-fhir/dsf-fhir-server/src/main/resources/fhir/db/db.changelog.xml

@@ -44,4 +44,5 @@
 	<include file="fhir/db/db.delete_duplicate_resources.changelog-1.6.1.xml" />
 
 	<include file="fhir/db/db.binaries.changelog-2.0.0.xml" />
+	<include file="fhir/db/db.binaries_lo_unlink_queue.changelog-2.0.0.xml" />
 </databaseChangeLog>

dsf-fhir/dsf-fhir-server/src/main/resources/fhir/db/trigger_functions/on_binaries_delete.sql

@@ -1,7 +1,11 @@
 CREATE OR REPLACE FUNCTION on_binaries_delete() RETURNS TRIGGER AS $$
 BEGIN
 	PERFORM on_resources_delete(OLD.binary_id);
-	PERFORM lo_unlink(OLD.binary_oid);
+
+	IF (OLD.binary_oid IS NOT NULL) THEN
+		INSERT INTO binaries_lo_unlink_queue (binary_oid) VALUES (OLD.binary_oid) ON CONFLICT DO NOTHING;
+	END IF;
+
 	RETURN OLD;
 END;
 $$ LANGUAGE PLPGSQL

dsf-fhir/dsf-fhir-server/src/main/resources/fhir/db/trigger_functions/on_binaries_lo_unlink_queue_delete.sql

@@ -0,0 +1,6 @@
+CREATE OR REPLACE FUNCTION on_binaries_lo_unlink_queue_delete() RETURNS TRIGGER AS $$
+BEGIN
+	PERFORM lo_unlink(OLD.binary_oid);
+	RETURN OLD;
+END;
+$$ LANGUAGE PLPGSQL

dsf-fhir/dsf-fhir-server/src/test/java/dev/dsf/fhir/integration/BinaryIntegrationTest.java

@@ -12,6 +12,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Date;
 import java.util.List;
@@ -2991,6 +2992,9 @@ public void testCreateReadDelete4GiB() throws Exception
 		long t1pd = System.currentTimeMillis();
 
 		logger.info("Permanent delete call finished in {} ms", (t1pd - t0pd));
+		
+		// wait for binaries_lo_unlink_queue delete to finish
+		Thread.sleep(Duration.ofSeconds(2));
 	}
 
 	@Test