Fix Shepherd/JDO connection leaks causing stuck dbconnections entries

src/main/java/org/ecocean/Encounter.java

@@ -4135,11 +4135,18 @@ public static void opensearchIndexPermissions() {
         OpenSearch os = new OpenSearch();
         Map<String, Set<String> > collab = new HashMap<String, Set<String> >();
         Map<String, String> usernameToId = new HashMap<String, String>();
+
+        // PHASE 1: load all DB data we need into in-memory structures, then close the Shepherd.
+        // The per-encounter OpenSearch HTTP calls in PHASE 2 below were previously made while
+        // this Shepherd's Postgres connection was still pinned, which on installs with hundreds
+        // of thousands of encounters could starve the JDO connection pool for tens of minutes
+        // and cause every concurrent request to queue at "begin" on dbconnections.jsp.
+        List<String[]> encounterRows = new ArrayList<String[]>();
         Shepherd myShepherd = new Shepherd("context0");
         myShepherd.setAction("Encounter.opensearchIndexPermissions");
         myShepherd.beginDBTransaction();
-        // it seems as though user.uuid is *required* so we can trust that
         try {
+            // it seems as though user.uuid is *required* so we can trust that
             for (User user : myShepherd.getUsersWithUsername()) {
                 usernameToId.put(user.getUsername(), user.getId());
                 List<Collaboration> collabsFor = Collaboration.collaborationsForUser(myShepherd,
@@ -4152,93 +4159,68 @@ public static void opensearchIndexPermissions() {
                     collab.get(user.getId()).add(col.getOtherUsername(user.getUsername()));
                 }
             }
+            Util.mark("perm: user build done", startT);
+            System.out.println("opensearchIndexPermissions(): " + usernameToId.size() +
+                " total users; " + collab.size() + " have active collab");
+
+            // we do not need full Encounter objects here to update index docs, so lets do this via sql/fields - much faster
+            String sql =
+                "SELECT \"CATALOGNUMBER\", \"SUBMITTERID\" FROM \"ENCOUNTER\" WHERE \"SUBMITTERID\" IS NOT NULL AND \"SUBMITTERID\" != '' AND \"SUBMITTERID\" != 'N/A' AND \"SUBMITTERID\" != 'public'";
+            Query q = null;
+            try {
+                q = myShepherd.getPM().newQuery("javax.jdo.query.SQL", sql);
+                List results = (List)q.execute();
+                Util.mark("perm: loading encs into memory, size=" + results.size(), startT);
+                Iterator it = results.iterator();
+                while (it.hasNext()) {
+                    Object[] row = (Object[])it.next();
+                    encounterRows.add(new String[] { (String)row[0], (String)row[1] });
+                }
+            } finally {
+                if (q != null) q.closeAll();
+            }
         } catch (Exception ex) {
+            System.out.println(
+                "opensearchIndexPermissions(): failed during DB load phase: " + ex);
             ex.printStackTrace();
+        } finally {
+            myShepherd.rollbackAndClose();
         }
-        Util.mark("perm: user build done", startT);
-        System.out.println("opensearchIndexPermissions(): " + usernameToId.size() +
-            " total users; " + collab.size() + " have active collab");
-        // now iterated over (non-public) encounters
+        Util.mark("perm: DB load done; shepherd closed", startT);
+
+        // PHASE 2: iterate the in-memory rows and update OpenSearch (no DB connection held).
         int encCount = 0;
-        org.json.JSONObject updateData = new org.json.JSONObject();
-        // we do not need full Encounter objects here to update index docs, so lets do this via sql/fields - much faster
-        String sql =
-            "SELECT \"CATALOGNUMBER\", \"SUBMITTERID\" FROM \"ENCOUNTER\" WHERE \"SUBMITTERID\" IS NOT NULL AND \"SUBMITTERID\" != '' AND \"SUBMITTERID\" != 'N/A' AND \"SUBMITTERID\" != 'public'";
-        Query q = null;
-        try {
-            q = myShepherd.getPM().newQuery("javax.jdo.query.SQL", sql);
-            List results = (List)q.execute();
-            Util.mark("perm: start encs, size=" + results.size(), startT);
-            Iterator it = results.iterator();
-            while (it.hasNext()) {
-                Object[] row = (Object[])it.next();
-                String id = (String)row[0];
-                String submitterId = (String)row[1];
-                org.json.JSONArray viewUsers = new org.json.JSONArray();
-                String uid = usernameToId.get(submitterId);
-                if (uid == null) {
-                    // see issue 939 for example :(
-                    System.out.println("opensearchIndexPermissions(): WARNING invalid username " +
-                        submitterId + " on enc " + id);
-                    continue;
-                }
-                encCount++;
-                if (encCount % 1000 == 0) Util.mark("enc[" + encCount + "]", startT);
-                // viewUsers.put(uid);  // we no longer do this as we use submitterUserId from regular indexing in query filter
-
-                // this first part asks the question: who is the owner of the Encounter collaborating with?
-                // Let those people see the encounter
-                // This ignores the one-way visibility of admins and orgAdmins
-                // the question is backwards: it asks: who can the owning user see?
-                // better to ask: who can see this Encounter by collaborating with its owner?
-                /*
-                   if (collab.containsKey(uid)) {
-                    for (String colUsername : collab.get(uid)) {
-                        String colId = usernameToId.get(colUsername);
-                        if (colId == null) {
-                            System.out.println(
-                                "opensearchIndexPermissions(): WARNING invalid username " +
-                                colUsername + " in collaboration with userId=" + uid);
-                            continue;
-                        }
-                        viewUsers.put(colId);
-                    }
-                   }*/
-
-                // better: ask the question, who else can see this encounter via collaboration?
-                // get the entry set for all collaborations
-                Set<String> uids = collab.keySet();
-                // iterate over the key set
-                Iterator<String> uidsIter = uids.iterator();
-                while (uidsIter.hasNext()) {
-                    // get the uid for the user of this entry
-                    String localUid = uidsIter.next();
-                    // get the list of usernames in this entry
-                    Set<String> localCollabs = collab.get(localUid);
-                    // evaluate if the submitterId (a username) of this encounter is in this list
-                    if (localCollabs.contains(submitterId)) {
-                        // if the submitterId is in the list, put the uid of the user in viewUsers for OpenSearch
-                        viewUsers.put(localUid);
-                    }
+        for (String[] row : encounterRows) {
+            String id = row[0];
+            String submitterId = row[1];
+            String uid = usernameToId.get(submitterId);
+            if (uid == null) {
+                // see issue 939 for example :(
+                System.out.println("opensearchIndexPermissions(): WARNING invalid username " +
+                    submitterId + " on enc " + id);
+                continue;
+            }
+            encCount++;
+            if (encCount % 1000 == 0) Util.mark("enc[" + encCount + "]", startT);
+
+            // ask the question, who else can see this encounter via collaboration?
+            org.json.JSONArray viewUsers = new org.json.JSONArray();
+            for (Map.Entry<String, Set<String> > entry : collab.entrySet()) {
+                if (entry.getValue().contains(submitterId)) {
+                    viewUsers.put(entry.getKey());
                 }
-                if (viewUsers.length() > 0) {
-                    updateData.put("viewUsers", viewUsers);
-                    try {
-                        os.indexUpdate("encounter", id, updateData);
-                    } catch (Exception ex) {
-                        // keeping this quiet cuz it can get noise while index builds
-                        // System.out.println("opensearchIndexPermissions(): WARNING failed to update viewUsers on enc " + enc.getId() + "; likely has not been indexed yet: " + ex);
-                    }
+            }
+            if (viewUsers.length() > 0) {
+                org.json.JSONObject updateData = new org.json.JSONObject();
+                updateData.put("viewUsers", viewUsers);
+                try {
+                    os.indexUpdate("encounter", id, updateData);
+                } catch (Exception ex) {
+                    // keeping this quiet cuz it can get noise while index builds
                 }
             }
-        } catch (Exception ex) {
-            System.out.println("opensearchIndexPermissions(): failed during encounter loop: " + ex);
-            ex.printStackTrace();
-        } finally {
-            if (q != null) q.closeAll();
         }
         Util.mark("perm: done encs", startT);
-        myShepherd.rollbackAndClose();
         System.out.println("opensearchIndexPermissions(): ...end [" + encCount + " encs; " +
             Math.round((System.currentTimeMillis() - startT) / 1000) + "sec]");
     }

src/main/java/org/ecocean/OpenSearch.java

@@ -171,27 +171,27 @@ public void run() {
     }
 
     private static void updatePermissionsIndex(String context) {
-        Shepherd myShepherd = new Shepherd(context);
-
-        myShepherd.setAction("OpenSearch.backgroundPermissions");
+        Shepherd myShepherd = null;
         try {
+            myShepherd = new Shepherd(context);
+            myShepherd.setAction("OpenSearch.backgroundPermissions");
             myShepherd.beginDBTransaction();
             System.out.println("OpenSearch background permissions running...");
             Encounter.opensearchIndexPermissionsBackground(myShepherd);
             System.out.println("OpenSearch background permissions finished.");
             myShepherd.commitDBTransaction(); // need commit since we might have changed SystemValues
-            myShepherd.closeDBTransaction();
         } catch (Exception ex) {
             ex.printStackTrace();
-            myShepherd.rollbackAndClose();
+        } finally {
+            if (myShepherd != null) myShepherd.rollbackAndClose();
         }
     }
 
     public static void updateEncounterIndexes(String context) {
-        Shepherd myShepherd = new Shepherd(context);
-
-        myShepherd.setAction("OpenSearch.backgroundIndexing");
+        Shepherd myShepherd = null;
         try {
+            myShepherd = new Shepherd(context);
+            myShepherd.setAction("OpenSearch.backgroundIndexing");
             myShepherd.beginDBTransaction();
             System.out.println("OpenSearch background indexing running...");
             Base.opensearchSyncIndex(myShepherd, Encounter.class, BACKGROUND_SLICE_SIZE);
@@ -203,7 +203,7 @@ public static void updateEncounterIndexes(String context) {
         } catch (Exception ex) {
             ex.printStackTrace();
         } finally {
-            myShepherd.rollbackAndClose();
+            if (myShepherd != null) myShepherd.rollbackAndClose();
             unsetActiveIndexingBackground();
         }
     }
@@ -894,17 +894,17 @@ public static void setPermissionsNeeded(Shepherd myShepherd, boolean value) {
     }
 
     public static void setPermissionsNeeded(boolean value) {
-        Shepherd myShepherd = new Shepherd("context0");
-
-        myShepherd.setAction("OpenSearch.setPermissionsNeeded");
-        myShepherd.beginDBTransaction();
+        Shepherd myShepherd = null;
         try {
+            myShepherd = new Shepherd("context0");
+            myShepherd.setAction("OpenSearch.setPermissionsNeeded");
+            myShepherd.beginDBTransaction();
             setPermissionsNeeded(myShepherd, value);
             myShepherd.commitDBTransaction();
-            myShepherd.closeDBTransaction();
         } catch (Exception ex) {
             ex.printStackTrace();
-            myShepherd.rollbackAndClose();
+        } finally {
+            if (myShepherd != null) myShepherd.rollbackAndClose();
         }
     }
 
@@ -1014,32 +1014,32 @@ public static void unsetActiveIndexingBackground() {
 
     static void setActive(String type) {
         // we want our own shepherd as the main shepherd may not persist this til later
-        Shepherd myShepherd = new Shepherd("context0");
-
-        myShepherd.setAction("OpenSearch.setActive");
-        myShepherd.beginDBTransaction();
+        Shepherd myShepherd = null;
         try {
+            myShepherd = new Shepherd("context0");
+            myShepherd.setAction("OpenSearch.setActive");
+            myShepherd.beginDBTransaction();
             SystemValue.set(myShepherd, type, true);
             myShepherd.commitDBTransaction();
-            myShepherd.closeDBTransaction();
         } catch (Exception ex) {
             ex.printStackTrace();
-            myShepherd.rollbackAndClose();
+        } finally {
+            if (myShepherd != null) myShepherd.rollbackAndClose();
         }
     }
 
     static void unsetActive(String type) {
-        Shepherd myShepherd = new Shepherd("context0");
-
-        myShepherd.setAction("OpenSearch.unsetActive");
-        myShepherd.beginDBTransaction();
+        Shepherd myShepherd = null;
         try {
+            myShepherd = new Shepherd("context0");
+            myShepherd.setAction("OpenSearch.unsetActive");
+            myShepherd.beginDBTransaction();
             SystemValue.set(myShepherd, type, false);
             myShepherd.commitDBTransaction();
-            myShepherd.closeDBTransaction();
         } catch (Exception ex) {
             ex.printStackTrace();
-            myShepherd.rollbackAndClose();
+        } finally {
+            if (myShepherd != null) myShepherd.rollbackAndClose();
         }
     }
 

src/main/java/org/ecocean/api/BulkImport.java

@@ -36,13 +36,13 @@ protected void doGet(HttpServletRequest request, HttpServletResponse response)
     throws ServletException, IOException {
         String context = ServletUtilities.getContext(request);
         int statusCode = 500;
-        Shepherd myShepherd = new Shepherd(context);
-
-        myShepherd.setAction("api.Bulk.doGet");
-        myShepherd.beginDBTransaction();
         JSONObject rtn = new JSONObject("{\"success\": false}");
+        Shepherd myShepherd = null;
 
         try {
+            myShepherd = new Shepherd(context);
+            myShepherd.setAction("api.Bulk.doGet");
+            myShepherd.beginDBTransaction();
             User currentUser = myShepherd.getUser(request);
             if (currentUser == null) {
                 response.setStatus(401);
@@ -121,7 +121,7 @@ protected void doGet(HttpServletRequest request, HttpServletResponse response)
         } catch (Exception ex) {
             ex.printStackTrace();
         } finally {
-            myShepherd.rollbackAndClose();
+            if (myShepherd != null) myShepherd.rollbackAndClose();
         }
         rtn.put("statusCode", statusCode);
         response.setStatus(statusCode);
@@ -144,12 +144,13 @@ protected void doPost(HttpServletRequest request, HttpServletResponse response)
         JSONObject encAssets = null;
         String dupId = null; // gets set as bulkImporId to be used in finally block
         long startProcess = System.currentTimeMillis();
-        Shepherd myShepherd = new Shepherd(context);
+        Shepherd myShepherd = null;
 
-        myShepherd.setAction("api.Bulk.doPost");
-        myShepherd.beginDBTransaction();
         long startTime = System.currentTimeMillis();
         try {
+            myShepherd = new Shepherd(context);
+            myShepherd.setAction("api.Bulk.doPost");
+            myShepherd.beginDBTransaction();
             User currentUser = myShepherd.getUser(request);
             if (currentUser == null) {
                 response.setStatus(401);
@@ -416,10 +417,11 @@ protected void doPost(HttpServletRequest request, HttpServletResponse response)
                     final boolean bgSkipDetection = skipDetection;
                     final boolean bgSkipIdentification = skipIdentification;
                     final String currentUsername = currentUser.getUsername();
+                    final String bgContext = myShepherd.getContext();
                     Runnable r = new Runnable() {
                         public void run() {
                             // make our background thread safely use our own Shepherd
-                            Shepherd bgShepherd = new Shepherd(myShepherd.getContext());
+                            Shepherd bgShepherd = new Shepherd(bgContext);
                             bgShepherd.setAction("api.Bulk.processBackground");
                             bgShepherd.beginDBTransaction();
 
@@ -618,12 +620,14 @@ public void run() {
             statusCode = 500;
             ex.printStackTrace();
         } finally {
-            if ((statusCode == 200) && !validateOnly) {
-                myShepherd.commitDBTransaction();
-            } else {
-                myShepherd.rollbackDBTransaction();
+            if (myShepherd != null) {
+                if ((statusCode == 200) && !validateOnly) {
+                    myShepherd.commitDBTransaction();
+                } else {
+                    myShepherd.rollbackDBTransaction();
+                }
+                myShepherd.closeDBTransaction();
             }
-            myShepherd.closeDBTransaction();
             if ((statusCode == 200) && !skipDetection)
                 initiateIA(dupId, skipIdentification, encAssets, matchingSetFilter);
         }