Session leak in set_tombstone
Motivation
----------
Currently the `core.replica.set_tombstone` method has a session leak, since it creates a new session with every method call. This breaks the session hand-through logic Rucio usually runs and needs to be fixed.


diff --git a/lib/rucio/core/replica.py b/lib/rucio/core/replica.py
index 0572d0146..a500fd7fb 100644
--- a/lib/rucio/core/replica.py
+++ b/lib/rucio/core/replica.py
@@ -61,7 +61,7 @@ from rucio.db.sqla import models
 from rucio.db.sqla.constants import (DIDType, ReplicaState, OBSOLETE, DIDAvailability,
                                      BadFilesStatus, RuleState, BadPFNStatus)
 from rucio.db.sqla.session import (read_session, stream_session, transactional_session,
-                                   DEFAULT_SCHEMA_NAME, get_engine)
+                                   DEFAULT_SCHEMA_NAME)
 from rucio.rse import rsemanager as rsemgr
 
 
@@ -2956,11 +2956,10 @@ def set_tombstone(rse_id, scope, name, tombstone=OBSOLETE, session=None):
     :param tombstone: the tombstone to set. Default is OBSOLETE
     :param session: database session in use.
     """
-    conn = get_engine().connect()
     stmt = update(models.RSEFileAssociation).where(and_(models.RSEFileAssociation.rse_id == rse_id, models.RSEFileAssociation.name == name, models.RSEFileAssociation.scope == scope,
                                                         ~session.query(models.ReplicaLock).filter_by(scope=scope, name=name, rse_id=rse_id).exists()))\\
                                             .values(tombstone=tombstone)
-    result = conn.execute(stmt)
+    result = session.execute(stmt)
     if not result.rowcount:
         try:
             session.query(models.RSEFileAssociation).filter_by(scope=scope, name=name, rse_id=rse_id).one()
diff --git a/lib/rucio/daemons/conveyor/finisher.py b/lib/rucio/daemons/conveyor/finisher.py
index c8fdce9fc..0a69bcde7 100644
--- a/lib/rucio/daemons/conveyor/finisher.py
+++ b/lib/rucio/daemons/conveyor/finisher.py
@@ -408,7 +408,7 @@ def __update_bulk_replicas(replicas, session=None):
     :returns commit_or_rollback:  Boolean.
     """
     try:
-        replica_core.update_replicas_states(replicas, nowait=True, add_tombstone=False, session=session)  # TODO: add_tombstone needs to be changed to True once set_tombstone is fixed
+        replica_core.update_replicas_states(replicas, nowait=True, add_tombstone=True, session=session)
     except ReplicaNotFound as error:
         logging.warn('Failed to bulk update replicas, will do it one by one: %s', str(error))
         raise ReplicaNotFound(error)
@@ -432,7 +432,7 @@ def __update_replica(replica, session=None):
     """
 
     try:
-        replica_core.update_replicas_states([replica], nowait=True, add_tombstone=False, session=session)  # TODO: add_tombstone needs to be changed to True once set_tombstone is fixed
+        replica_core.update_replicas_states([replica], nowait=True, add_tombstone=True, session=session)
         if not replica['archived']:
             request_core.archive_request(replica['request_id'], session=session)
         logging.info("HANDLED REQUEST %s DID %s:%s AT RSE %s STATE %s", replica['request_id'], replica['scope'], replica['name'], replica['rse_id'], str(replica['state']))
['lib/rucio/daemons/conveyor/finisher.py', 'lib/rucio/core/replica.py']