fix(FeedUpdater): Prevent incorrect published status update on server startup.

Author: binh-dam-ibigroup

src/main/java/com/conveyal/datatools/manager/jobs/FeedUpdater.java

@@ -158,6 +158,7 @@ public Map<String, String> checkForUpdatedFeeds() {
             String keyName = objSummary.getKey();
             LOG.debug("{} etag = {}", keyName, eTag);
 
+            // Don't add object if it is a dir
             if (keyName.equals(bucketFolder)) continue;
             String filename = keyName.split("/")[1];
             String feedId = filename.replace(".zip", "");
@@ -166,16 +167,20 @@ public Map<String, String> checkForUpdatedFeeds() {
                 LOG.error("No feed source found for feed ID {}", feedId);
                 continue;
             }
+            // Skip object if the filename is null
+            if ("null".equals(feedId)) continue;
 
-            if (shouldMarkFeedAsProcessed(eTag, feedSource)) {
-                // Don't add object if it is a dir
-                // Skip object if the filename is null
-                if ("null".equals(feedId)) continue;
+            FeedVersion latestVersionSentForPublishing = getLatestVersionSentForPublishing(feedId, feedSource);
+            if (shouldMarkFeedAsProcessed(eTag, latestVersionSentForPublishing)) {
                 try {
-                    LOG.info("New version found for {} at s3://{}/{}. ETag = {}.", feedId, feedBucket, keyName, eTag);
-                    updatePublishedFeedVersion(feedId, feedSource);
-                    // TODO: Explore if MD5 checksum can be used to find matching feed version.
-                    // findMatchingFeedVersion(md5, feedId, feedSource);
+                    // Don't mark a feed version as published if previous published version is before sentToExternalPublisher.
+                    if (!objSummary.getLastModified().before(latestVersionSentForPublishing.sentToExternalPublisher)) {
+                        LOG.info("New version found for {} at s3://{}/{}. ETag = {}.", feedId, feedBucket, keyName, eTag);
+                        updatePublishedFeedVersion(feedId, latestVersionSentForPublishing);
+                        // TODO: Explore if MD5 checksum can be used to find matching feed version.
+                        // findMatchingFeedVersion(md5, feedId, feedSource);
+                    }
+
                 } catch (Exception e) {
                     LOG.warn("Could not load feed " + keyName, e);
                 } finally {
@@ -200,7 +205,7 @@ private FeedSource getFeedSource(String feedId) {
             and(eq("value", feedId), eq("name", AGENCY_ID_FIELDNAME))
         );
         if (properties.size() > 1) {
-            LOG.warn("Found multiple feed sources for {}: {}",
+            LOG.warn("Found multiple feed sources for {}: {}. The published status on some feed versions will be incorrect.",
                 feedId,
                     properties.stream().map(p -> p.feedSourceId).collect(Collectors.joining(",")));
         }
@@ -216,22 +221,20 @@ private FeedSource getFeedSource(String feedId) {
     /**
      * @return true if the feed with the corresponding etag should be mark as processed, false otherwise.
      */
-    private boolean shouldMarkFeedAsProcessed(String eTag, FeedSource feedSource) {
+    private boolean shouldMarkFeedAsProcessed(String eTag, FeedVersion publishedVersion) {
         if (eTagForFeed.containsValue(eTag)) return false;
-
-        FeedVersion publishedVersion = getLatestPublishedVersion(feedSource);
         if (publishedVersion == null) return false;
+
         return versionsToMarkAsProcessed.contains(publishedVersion.id);
     }
 
     /**
      * Update the published feed version for the feed source.
      * @param feedId the unique ID used by MTC to identify a feed source
-     * @param feedSource the feed source for which a newly published version should be registered
+     * @param publishedVersion the feed version to be registered
      */
-    private void updatePublishedFeedVersion(String feedId, FeedSource feedSource) {
+    private void updatePublishedFeedVersion(String feedId, FeedVersion publishedVersion) {
         try {
-            FeedVersion publishedVersion = getLatestPublishedVersion(feedSource);
             if (publishedVersion != null) {
                 if (publishedVersion.sentToExternalPublisher == null) {
                     LOG.warn("Not updating published version for {} (version was never sent to external publisher)", feedId);
@@ -240,13 +243,18 @@ private void updatePublishedFeedVersion(String feedId, FeedSource feedSource) {
                 // Set published namespace to the feed version and set the processedByExternalPublisher timestamp.
                 LOG.info("Latest published version (sent at {}) for {} is {}", publishedVersion.sentToExternalPublisher, feedId, publishedVersion.id);
                 Persistence.feedVersions.updateField(publishedVersion.id, PROCESSED_BY_EXTERNAL_PUBLISHER_FIELD, new Date());
-                Persistence.feedSources.updateField(feedSource.id, "publishedVersionId", publishedVersion.namespace);
+                Persistence.feedSources.updateField(publishedVersion.feedSourceId, "publishedVersionId", publishedVersion.namespace);
             } else {
-                LOG.error("No published versions found for {} ({} id={})", feedId, feedSource.name, feedSource.id);
+                LOG.error(
+                    "No published versions found for {} ({} id={})",
+                    feedId,
+                    publishedVersion.parentFeedSource().name,
+                    publishedVersion.feedSourceId
+                );
             }
         } catch (Exception e) {
             e.printStackTrace();
-            LOG.error("Error encountered while checking for latest published version for {}", feedId);
+            LOG.error("Error encountered while updating the latest published version for {}", feedId);
         }
     }
 
@@ -256,13 +264,19 @@ private void updatePublishedFeedVersion(String feedId, FeedSource feedSource) {
      * could be that more than one versions were recently "published" and the latest published version was a bad
      * feed that failed processing by RTD.
      */
-    private static FeedVersion getLatestPublishedVersion(FeedSource feedSource) {
-        // Collect the feed versions for the feed source.
-        Collection<FeedVersion> versions = feedSource.retrieveFeedVersions();
-        Optional<FeedVersion> lastPublishedVersionCandidate = versions
-            .stream()
-            .min(Comparator.comparing(v -> v.sentToExternalPublisher, Comparator.nullsLast(Comparator.reverseOrder())));
-        return lastPublishedVersionCandidate.orElse(null);
+    private static FeedVersion getLatestVersionSentForPublishing(String feedId, FeedSource feedSource) {
+        try {
+            // Collect the feed versions for the feed source.
+            Collection<FeedVersion> versions = feedSource.retrieveFeedVersions();
+            Optional<FeedVersion> lastPublishedVersionCandidate = versions
+                .stream()
+                .min(Comparator.comparing(v -> v.sentToExternalPublisher, Comparator.nullsLast(Comparator.reverseOrder())));
+            return lastPublishedVersionCandidate.orElse(null);
+        } catch (Exception e) {
+            e.printStackTrace();
+            LOG.error("Error encountered while checking for latest published version for {}", feedId);
+            return null;
+        }
     }
 
     /**

src/test/java/com/conveyal/datatools/manager/jobs/AutoPublishJobTest.java

@@ -12,6 +12,7 @@
 import com.google.common.collect.Lists;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
@@ -163,7 +164,7 @@ void shouldUpdateFeedInfoAfterPublishComplete(String agencyId, boolean isUnknown
         assertTrue(etags.isEmpty());
 
         // Simulate completion of feed publishing.
-        completedFeedRetriever.isPublishingComplete = true;
+        completedFeedRetriever.makePublished();
 
         // The etags should contain the id of the agency.
         // If a feed has been republished since last check, it will have a new etag/file hash,
@@ -209,13 +210,62 @@ private static Stream<Arguments> createUpdateFeedInfoCases() {
         );
     }
 
+    /**
+     * This test ensures that, upon server startup,
+     * feeds that meet all these criteria should not be updated/marked as published:
+     * - the feed has been sent to publisher (RTD),
+     * - the publisher has not published the feed,
+     * - a previous version of the feed was already published.
+     */
+    @Test
+    void shouldNotUpdateFromAPreviouslyPublishedVersionOnStartup() {
+        final int TWO_DAYS_MILLIS = 48 * 3600000;
+
+        // Set up a test FeedUpdater instance that fakes an external published date in the past.
+        TestCompletedFeedRetriever completedFeedRetriever = new TestCompletedFeedRetriever(TEST_AGENCY);
+        FeedUpdater feedUpdater = FeedUpdater.createForTest(completedFeedRetriever);
+        completedFeedRetriever.makePublished(new Date(System.currentTimeMillis() - TWO_DAYS_MILLIS));
+
+        // Add the version to the feed source, with
+        // sentToExternalPublisher set to a date after a previous publish date.
+        FeedVersion createdVersion = createFeedVersionFromGtfsZip(feedSource, "bart_new_lite.zip");
+        createdVersion.sentToExternalPublisher = new Date();
+        Persistence.feedVersions.replace(createdVersion.id, createdVersion);
+
+        // The list of feeds processed externally (completed) should contain an entry for the agency we want.
+        Map<String, String> etags = feedUpdater.checkForUpdatedFeeds();
+        assertNotNull(etags.get(TEST_AGENCY));
+
+        // Make sure that the feed remains unpublished.
+        FeedVersion updatedFeedVersion = Persistence.feedVersions.getById(createdVersion.id);
+        assertNull(updatedFeedVersion.processedByExternalPublisher);
+
+        // Now perform publishing.
+        AutoPublishJob autoPublishJob = new AutoPublishJob(feedSource, user);
+        autoPublishJob.run();
+        assertFalse(autoPublishJob.status.error);
+
+        // Simulate another publishing process
+        completedFeedRetriever.makePublished(new Date());
+
+        // The list of feeds processed externally (completed) should contain an entry for the agency we want.
+        Map<String, String> etagsAfter = feedUpdater.checkForUpdatedFeeds();
+        assertNotNull(etagsAfter.get(TEST_AGENCY));
+
+        // The feed should be published.
+        FeedVersion publishedFeedVersion = Persistence.feedVersions.getById(createdVersion.id);
+        assertNotNull(publishedFeedVersion.processedByExternalPublisher);
+
+    }
+
     /**
      * Mocks the results of an {@link S3ObjectSummary} retrieval before/after the
      * external MTC publishing process is complete.
      */
     private static class TestCompletedFeedRetriever implements FeedUpdater.CompletedFeedRetriever {
         private final String agencyId;
-        public boolean isPublishingComplete;
+        private boolean isPublishingComplete;
+        private Date publishDate;
 
         public TestCompletedFeedRetriever(String agencyId) {
             this.agencyId = agencyId;
@@ -229,8 +279,18 @@ public List<S3ObjectSummary> retrieveCompletedFeeds() {
                 S3ObjectSummary objSummary = new S3ObjectSummary();
                 objSummary.setETag("test-etag");
                 objSummary.setKey(String.format("%s/%s", TEST_COMPLETED_FOLDER, agencyId));
+                objSummary.setLastModified(publishDate);
                 return Lists.newArrayList(objSummary);
             }
         }
+
+        public void makePublished() {
+            makePublished(new Date());
+        }
+
+        public void makePublished(Date publishDate) {
+            isPublishingComplete = true;
+            this.publishDate = publishDate;
+        }
     }
 }