In IntervalScheduler, keep Future-returning ticks overlap-free #8402
Conversation
📝 WalkthroughWalkthroughThe changes update several controller and service classes to improve type safety, clarity, and logging. Method signatures for tick-like operations and Changes
Possibly related PRs
Suggested labels
Suggested reviewers
Poem
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (5)
webknossos-datastore/app/com/scalableminds/webknossos/datastore/helpers/IntervalScheduler.scala (2)
20-20: NewtickerEnabledflag.
The toggle is helpful for conditionally disabling the scheduler. Consider making it configurable if runtime control is needed.
45-46: Clock-based logic may be affected by system time changes.
Relying on wall-clock time could behave unexpectedly if the system time is altered. If needed, consider using a monotonic clock or verifying time correctness.app/models/storage/UsedStorageService.scala (1)
44-53: Consider adding overlap protection.The removal of the
isRunningcheck could potentially lead to overlapping scans if the Fox chain takes longer than the ticker interval. While the Fox monad handles asynchronous operations, it doesn't prevent a new tick from starting while the previous one is still running.Consider implementing overlap protection using a semaphore or mutex to ensure only one scan runs at a time:
private val scanLock = new java.util.concurrent.Semaphore(1) override protected def tick(): Fox[Unit] = Fox.guard(scanLock.tryAcquire()) { try { for { organizations <- organizationDAO.findNotRecentlyScanned(...) dataStores <- dataStoreDAO.findAllWithStorageReporting _ <- Fox.serialCombined(organizations)(...) } yield () } finally { scanLock.release() } }webknossos-datastore/app/com/scalableminds/webknossos/datastore/services/DSRemoteWebknossosClient.scala (1)
63-63: Consider using Fox.void for cleaner code.The current implementation can be simplified using
Fox.voidinstead ofmap(_ => ()).- def tick(): Fox[Unit] = reportStatus().map(_ => ()) + def tick(): Fox[Unit] = Fox.void(reportStatus())webknossos-datastore/app/com/scalableminds/webknossos/datastore/services/DataSourceService.scala (1)
51-56: Simplify the tick implementation.While the change to
Fox[Unit]is good for making ticks overlap-free, the implementation can be simplified.Consider this simpler implementation:
-def tick(): Fox[Unit] = - for { - _ <- checkInbox(verbose = inboxCheckVerboseCounter == 0) - _ = inboxCheckVerboseCounter += 1 - _ = if (inboxCheckVerboseCounter >= 10) inboxCheckVerboseCounter = 0 - } yield () +def tick(): Fox[Unit] = { + val verbose = inboxCheckVerboseCounter == 0 + inboxCheckVerboseCounter = (inboxCheckVerboseCounter + 1) % 10 + checkInbox(verbose) +}
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (9)
app/controllers/WKRemoteDataStoreController.scala(3 hunks)app/mail/MailchimpTicker.scala(2 hunks)app/models/annotation/AnnotationMutexService.scala(2 hunks)app/models/job/Worker.scala(2 hunks)app/models/storage/UsedStorageService.scala(2 hunks)webknossos-datastore/app/com/scalableminds/webknossos/datastore/helpers/IntervalScheduler.scala(1 hunks)webknossos-datastore/app/com/scalableminds/webknossos/datastore/services/DSRemoteWebknossosClient.scala(2 hunks)webknossos-datastore/app/com/scalableminds/webknossos/datastore/services/DataSourceService.scala(2 hunks)webknossos-datastore/app/com/scalableminds/webknossos/datastore/services/DatasetErrorLoggingService.scala(2 hunks)
🔇 Additional comments (20)
webknossos-datastore/app/com/scalableminds/webknossos/datastore/helpers/IntervalScheduler.scala (10)
3-3: ImportingFoxis consistent with the new return type.
No concerns here; this aligns nicely with the updatedtick()signature.
7-7: ImportingAtomicLongfor concurrency tracking.
This is a reasonable approach to managing shared mutable state for time tracking.
10-10: ImportingInstantfor timestamp handling.
This is a clear way to handle time references and epoch milliseconds.
18-18: RenamingsystemtoactorSystem.
This improves clarity; no issues.
26-26: Changingtick()fromUnittoFox[_].
This is a good step toward asynchronous/functional handling. However, usingFox[Unit]instead ofFox[_]might improve type clarity if no generic variance is required.
28-30: Introducing concurrency fields.
innerTickerInterval,lastCompletionTimeMillis, andisRunningcollectively manage timing and concurrency. This setup is logically consistent.
49-51: Conditional cancellation ofscheduled.
Cleaning up existing tasks is important and this approach looks sound.
55-55:schedulednow anOption[Cancellable].
UsingOptionis a neat way to handle the absence of a scheduled task. Ensure concurrency safety if manipulated from multiple threads.
58-58: Lifecycle’saddStopHook.
Tidy integration to ensure the scheduler stops cleanly.
60-67: Initializing the ticker only iftickerEnabled.
Logic is straightforward: checking intervals, throwing an exception if too short, and scheduling. Looks good.app/mail/MailchimpTicker.scala (2)
19-19: Renaming constructor parameter toactorSystem.
This improves consistency with the rest of the updated codebase.
31-31: Changingtick()return type toFox[Unit].
Adopting aFoxreturn type for asynchronous handling is consistent with the new scheduling pattern. Looks good.webknossos-datastore/app/com/scalableminds/webknossos/datastore/services/DatasetErrorLoggingService.scala (2)
21-21: Renaming toactorSystem.
No concerns; aligns naming with other services.
50-50: Convertingtick()fromUnittoFox[Unit].
ReturningFox[Unit]is coherent with the broader asynchronous approach. ClearingrecentErrorsis unlikely to fail, so this seems sufficient.app/models/annotation/AnnotationMutexService.scala (1)
27-27: LGTM! The changes improve error handling.The conversion to
Fox[Unit]return type is well-implemented, properly handling potential database operation failures when cleaning up expired mutexes.Also applies to: 37-40
app/models/job/Worker.scala (1)
105-105: LGTM! The changes align with async worker monitoring.The conversion to
Fox[Unit]return type is appropriate for the worker liveness monitoring, properly handling potential database operation failures.Also applies to: 114-118
webknossos-datastore/app/com/scalableminds/webknossos/datastore/services/DataSourceService.scala (2)
33-33: LGTM! Improved parameter naming.The rename from
systemtoactorSystemenhances code clarity by being more specific about the parameter's purpose.
39-40: LGTM! Improved ticker configuration.The changes enhance clarity and stability:
- Renamed
enabledtotickerEnabledfor better consistency- Fixed interval of 2 seconds helps prevent overlapping ticks, addressing the PostgreSQL overload issue
app/controllers/WKRemoteDataStoreController.scala (2)
5-5: LGTM! Added necessary imports.The imports support the new performance monitoring feature.
Also applies to: 36-36
189-204: LGTM! Improved type safety and added performance monitoring.The changes enhance the method in several ways:
- Type-safe request body validation using
validateJson[List[InboxDataSource]]- Performance monitoring to identify slow updates (>30s)
- Cleaner implementation by removing explicit JSON validation
| private def innerTick: Runnable = () => { | ||
| if (lastCompletionIsLongEnoughPast) { | ||
| if (isRunning.compareAndSet(false, true)) { | ||
| for { | ||
| _ <- tick().futureBox | ||
| _ = lastCompletionTimeMillis.set(Instant.now.epochMillis) | ||
| _ = isRunning.set(false) | ||
| } yield () | ||
| } | ||
| } | ||
| () |
There was a problem hiding this comment.
Ensure isRunning is released on exceptions.
If tick().futureBox fails, isRunning remains true, causing a permanent block. Use a recover or similar mechanism to reset isRunning in error cases.
Example fix:
for {
_ <- tick().futureBox.recover {
+ case e =>
+ isRunning.set(false)
+ throw e
}
_ = lastCompletionTimeMillis.set(Instant.now.epochMillis)
_ = isRunning.set(false)
} yield ()📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| private def innerTick: Runnable = () => { | |
| if (lastCompletionIsLongEnoughPast) { | |
| if (isRunning.compareAndSet(false, true)) { | |
| for { | |
| _ <- tick().futureBox | |
| _ = lastCompletionTimeMillis.set(Instant.now.epochMillis) | |
| _ = isRunning.set(false) | |
| } yield () | |
| } | |
| } | |
| () | |
| private def innerTick: Runnable = () => { | |
| if (lastCompletionIsLongEnoughPast) { | |
| if (isRunning.compareAndSet(false, true)) { | |
| for { | |
| - _ <- tick().futureBox | |
| + _ <- tick().futureBox.recover { | |
| + case e => | |
| + isRunning.set(false) | |
| + throw e | |
| + } | |
| _ = lastCompletionTimeMillis.set(Instant.now.epochMillis) | |
| _ = isRunning.set(false) | |
| } yield () | |
| } | |
| } | |
| () | |
| } |
There was a problem hiding this comment.
Thanks for this PR. 🎉
Hopefully this fixes the recent errors 🤞
As you renamed the system: ActorSystem injects, could you maybe do this as well for the other cases which do not use the interval ticking? I found a few other occurences which still use system: ActorSystem.
Besides this and my comment everything looks good and works 🎉
| private val lastCompletionTimeMillis = new AtomicLong(0) | ||
| private val isRunning = new java.util.concurrent.atomic.AtomicBoolean(false) | ||
|
|
There was a problem hiding this comment.
Pretty nitpicky but this is a little inconsistent here. In the "header part" you import AtomicLong but not AtomicBoolean which you import inline here. I'd prefer to also import the AtomicBoolean in the header together with the AtomicLong.
There was a problem hiding this comment.
Actionable comments posted: 0
♻️ Duplicate comments (1)
webknossos-datastore/app/com/scalableminds/webknossos/datastore/helpers/IntervalScheduler.scala (1)
32-43:⚠️ Potential issueAdd error handling to ensure
isRunningis released on exceptions.The current implementation might leave
isRunningin a locked state iftick().futureBoxfails.Apply this diff to ensure proper cleanup:
private def innerTick: Runnable = () => { if (lastCompletionIsLongEnoughPast) { if (isRunning.compareAndSet(false, true)) { for { - _ <- tick().futureBox + _ <- tick().futureBox.recover { + case e => + isRunning.set(false) + throw e + } _ = lastCompletionTimeMillis.set(Instant.now.epochMillis) _ = isRunning.set(false) } yield () } } () }
🧹 Nitpick comments (3)
app/controllers/WKRemoteDataStoreController.scala (1)
189-204: Excellent improvements to type safety and monitoring!The changes enhance the code in several ways:
- Type-safe request body validation using
List[InboxDataSource]- Performance monitoring for long-running updates (>30s)
- Clear logging of active/inactive dataset counts
Consider adding metrics collection for the update duration to track performance trends over time. This would help identify gradual degradation in update performance.
webknossos-datastore/app/com/scalableminds/webknossos/datastore/helpers/IntervalScheduler.scala (2)
48-54: Consider adding synchronization to the stop method.The stop method might have a race condition when checking and updating the
scheduledvariable. Multiple concurrent calls to stop could lead to inconsistent state.Consider this improvement:
- private def stop(): Future[Unit] = { - if (scheduled.isDefined) { - scheduled.foreach(_.cancel()) - scheduled = None - } - Future.successful(()) - } + private def stop(): Future[Unit] = synchronized { + scheduled.foreach(_.cancel()) + scheduled = None + Future.successful(()) + }
61-64: Consider adding validation for negative intervals.While the code checks if the interval is too small, it should also validate that it's not negative.
Consider this addition:
if (tickerInterval < innerTickerInterval) { throw new IllegalArgumentException( s"IntervalScheduler was initialized with interval $tickerInterval. Only intervals larger than the inner ticker interval $innerTickerInterval are supported.") } +if (tickerInterval.toMillis < 0) { + throw new IllegalArgumentException( + s"IntervalScheduler was initialized with negative interval $tickerInterval.") +}
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (9)
app/controllers/WKRemoteDataStoreController.scala(3 hunks)app/models/voxelytics/LokiClient.scala(2 hunks)webknossos-datastore/app/com/scalableminds/webknossos/datastore/DataStoreModule.scala(1 hunks)webknossos-datastore/app/com/scalableminds/webknossos/datastore/helpers/IntervalScheduler.scala(1 hunks)webknossos-datastore/app/com/scalableminds/webknossos/datastore/services/DataSourceRepository.scala(1 hunks)webknossos-datastore/app/com/scalableminds/webknossos/datastore/services/DataSourceService.scala(2 hunks)webknossos-tracingstore/app/com/scalableminds/webknossos/tracingstore/TracingStoreModule.scala(1 hunks)webknossos-tracingstore/app/com/scalableminds/webknossos/tracingstore/tracings/TemporaryTracingStore.scala(1 hunks)webknossos-tracingstore/app/com/scalableminds/webknossos/tracingstore/tracings/TracingDataStore.scala(1 hunks)
✅ Files skipped from review due to trivial changes (4)
- webknossos-tracingstore/app/com/scalableminds/webknossos/tracingstore/tracings/TracingDataStore.scala
- webknossos-tracingstore/app/com/scalableminds/webknossos/tracingstore/tracings/TemporaryTracingStore.scala
- webknossos-datastore/app/com/scalableminds/webknossos/datastore/services/DataSourceRepository.scala
- webknossos-tracingstore/app/com/scalableminds/webknossos/tracingstore/TracingStoreModule.scala
🚧 Files skipped from review as they are similar to previous changes (1)
- webknossos-datastore/app/com/scalableminds/webknossos/datastore/services/DataSourceService.scala
⏰ Context from checks skipped due to timeout of 90000ms (1)
- GitHub Check: circleci_build
🔇 Additional comments (5)
webknossos-datastore/app/com/scalableminds/webknossos/datastore/DataStoreModule.scala (2)
12-12: LGTM! Improved encapsulation and naming clarity.The renaming from
systemtoactorSystemand making it private improves code clarity and encapsulation. This change aligns with similar renaming across other files in the codebase.
17-17: LGTM! Binding correctly updated.The binding has been correctly updated to use the renamed
actorSystemvariable while maintaining the same functionality.app/models/voxelytics/LokiClient.scala (1)
23-23: LGTM! Parameter renaming improves clarity.The renaming of
systemtoactorSystemenhances code readability by making the parameter's purpose more explicit.Also applies to: 46-46
app/controllers/WKRemoteDataStoreController.scala (1)
5-5: LGTM! Import statements support new functionality.The added imports enable timing measurement and duration specification for monitoring update operations.
Also applies to: 36-36
webknossos-datastore/app/com/scalableminds/webknossos/datastore/helpers/IntervalScheduler.scala (1)
18-29: LGTM! The changes improve type safety and scheduling control.The renaming of
systemtoactorSystemimproves clarity, and the introduction ofFox[_]return type enables proper handling of asynchronous operations. TheinnerTickerIntervalof 100ms provides a good safety buffer between ticks.
Before this PR, the tick functions returned Unit, meaning that if they started asynchronous work, the IntervalScheduler couldn’t wait for it to be completed.
Now, tick returns Fox[Unit]. The IntervalScheduler runs with an internal interval of 100ms and checks if the time since the last tick’s completion is long enough to start a new one. That means the real interval will be a little higher than the specified one, because it includes the time to compute the tick (plus the time until the next internal 100ms-spaced tick).
Steps to test:
Issues: