iosource/Manager: Honor manage_lifetime and dont_count for short-lived IO sources

If an IO source is registered and becomes dry at runtime, the IO
manager would not honor its manage_lifetime or dont_count attribute
during collection, resulting in memory leaks.

This probably hasn't mattered so far as there's no IO sources registered
in-tree at runtime using manage_lifetime=true.

src/iosource/Manager.cc

@@ -104,6 +104,22 @@ void Manager::Wakeup(std::string_view where) {
         wakeup->Ping(where);
 }
 
+void Manager::ReapSource(Source* src) {
+    auto* iosource = src->src;
+    assert(! iosource->IsOpen());
+
+    DBG_LOG(DBG_MAINLOOP, "Reaping %s", src->src->Tag());
+    iosource->Done();
+
+    if ( src->manage_lifetime )
+        delete iosource;
+
+    if ( src->dont_count )
+        dont_counts--;
+
+    delete src;
+}
+
 void Manager::FindReadySources(ReadySources* ready) {
     ready->clear();
 
@@ -111,8 +127,7 @@ void Manager::FindReadySources(ReadySources* ready) {
     // remove at most one each time.
     for ( SourceList::iterator i = sources.begin(); i != sources.end(); ++i )
         if ( ! (*i)->src->IsOpen() ) {
-            (*i)->src->Done();
+            ReapSource(*i);
-            delete *i;
             sources.erase(i);
             break;
         }

src/iosource/Manager.h

@@ -143,6 +143,15 @@ public:
     void Wakeup(std::string_view where);
 
 private:
+    /**
+     * Internal data structure for managing registered IOSources.
+     */
+    struct Source {
+        IOSource* src = nullptr;
+        bool dont_count = false;
+        bool manage_lifetime = false;
+    };
+
     /**
      * Calls the appropriate poll method to gather a set of IOSources that are
      * ready for processing.
@@ -170,6 +179,19 @@ private:
 
     void RemoveAll();
 
+    /**
+     * Reap a closed IO source.
+     *
+     * Reaping involves calling IOSource::Done() on the underlying IOSource,
+     * freeing it if Source.manage_lifetime is \c true, updating \c dont_counts
+     * and freeing \a src, making it invalid.
+     *
+     * The caller ensures \a src is removed from Manager.sources.
+     *
+     * @param src The source to reap.
+     */
+    void ReapSource(Source* src);
+
     class WakeupHandler final : public IOSource {
     public:
         WakeupHandler();
@@ -192,12 +214,6 @@ private:
         zeek::detail::Flare flare;
     };
 
-    struct Source {
-        IOSource* src = nullptr;
-        bool dont_count = false;
-        bool manage_lifetime = false;
-    };
-
     using SourceList = std::vector<Source*>;
     SourceList sources;