srm: Lock job while saving to create consistent persistent state

Motivation:

We have observed incosistencies in the srm database content in which the
request state was different than the last logged transition.

Modification:

Place a read lock on the job to get a consistent view of the job's state.

Result:

Hopefully a more consistent srm database.

Target: trunk
Request: 2.14
Request: 2.13
Require-notes: yes
Require-book: no
Acked-by: Paul Millar <paul.millar@desy.de>
Acked-by: Dmitry Litvintsev <litvinse@fnal.gov>
Patch: https://rb.dcache.org/r/8890/
(cherry picked from commit b77d532)

modules/srm-server/src/main/java/org/dcache/srm/request/sql/DatabaseJobStorage.java

@@ -362,44 +362,6 @@ private J getJob(Connection _con, ResultSet set) throws SQLException
         return job;
     }
 
-    private int updateJob(Connection connection, Job job) throws SQLException
-    {
-        PreparedStatement updateStatement = null;
-        try {
-            job.rlock();
-            try {
-                updateStatement = getUpdateStatement(connection, job);
-            } finally {
-                job.runlock();
-            }
-            return updateStatement.executeUpdate();
-        } finally {
-            SqlHelper.tryToClose(updateStatement);
-        }
-    }
-
-    private void createJob(Connection connection, Job job) throws SQLException
-    {
-        PreparedStatement createStatement = null;
-        PreparedStatement batchCreateStatement = null;
-        try {
-            job.rlock();
-            try {
-                createStatement = getCreateStatement(connection, job);
-                batchCreateStatement = getBatchCreateStatement(connection, job);
-            } finally {
-                job.runlock();
-            }
-            createStatement.executeUpdate();
-            if (batchCreateStatement != null) {
-                batchCreateStatement.executeBatch();
-            }
-        } finally {
-            SqlHelper.tryToClose(createStatement);
-            SqlHelper.tryToClose(batchCreateStatement);
-        }
-    }
-
     private void saveHistory(Connection connection, Job job,
                              List<Job.JobHistory> history) throws SQLException
     {
@@ -436,18 +398,41 @@ private List<Job.JobHistory> getJobHistoriesToSave(Job job)
     @Override
     public void saveJob(final Job job, boolean force) throws DataAccessException
     {
-        final List<Job.JobHistory> history = getJobHistoriesToSave(job);
-        transactionTemplate.execute(status -> jdbcTemplate.execute((Connection con) -> {
-            int rowCount = updateJob(con, job);
-            if (rowCount == 0) {
-                createJob(con, job);
-            }
-            if (!history.isEmpty()) {
-                saveHistory(con, job, history);
-            }
-            return null;
-        }));
-        markHistoryAsSaved(history);
+        List<Job.JobHistory> savedHistory =
+                transactionTemplate.execute(status -> jdbcTemplate.execute((Connection con) -> {
+                    List<Job.JobHistory> history;
+                    PreparedStatement updateStatement = null;
+                    PreparedStatement createStatement = null;
+                    PreparedStatement batchCreateStatement = null;
+                    try {
+                        job.rlock();
+                        try {
+                            history = getJobHistoriesToSave(job);
+                            updateStatement = getUpdateStatement(con, job);
+                            createStatement = getCreateStatement(con, job);
+                            batchCreateStatement = getBatchCreateStatement(con, job);
+                        } finally {
+                            job.runlock();
+                        }
+
+                        int rowCount = updateStatement.executeUpdate();
+                        if (rowCount == 0) {
+                            createStatement.executeUpdate();
+                            if (batchCreateStatement != null) {
+                                batchCreateStatement.executeBatch();
+                            }
+                        }
+                        if (!history.isEmpty()) {
+                            saveHistory(con, job, history);
+                        }
+                    } finally {
+                        SqlHelper.tryToClose(createStatement);
+                        SqlHelper.tryToClose(batchCreateStatement);
+                        SqlHelper.tryToClose(updateStatement);
+                    }
+                    return history;
+                }));
+        markHistoryAsSaved(savedHistory);
     }
 
     protected PreparedStatement getBatchCreateStatement(Connection connection, Job job)