From 1164e7a3531732dd2cb7457fe93e87d403c8efb8 Mon Sep 17 00:00:00 2001
From: Seongmin Lee <git@boltless.me>
Date: Sat, 04 Apr 2026 17:04:07 +0000
Subject: [PATCH] knotmirror/knotstream: use long-live context for subscription

Signed-off-by: Seongmin Lee <git@boltless.me>
---
 knotmirror/knotstream/knotstream.go | 14 ++++++++++++--
 knotmirror/knotstream/slurper.go | 7 ++++++-
 2 file(s) changed, 18 insertion(s)(+), 3 deletion(s)(-)

diff --git a/knotmirror/knotstream/knotstream.go b/knotmirror/knotstream/knotstream.go
--- a/knotmirror/knotstream/knotstream.go
+++ b/knotmirror/knotstream/knotstream.go
@@ -66,7 +66,17 @@
 	if host.Status == models.HostStatusBanned {
 		return fmt.Errorf("cannot subscribe to banned knot")
 	}
-	return s.slurper.Subscribe(ctx, *host)
+	// `Subscribe` expects long-living context
+	if err := s.slurper.Subscribe(*host); err != nil {
+		return fmt.Errorf("slurper: %w", err)
+	}
+
+	host.Status = models.HostStatusActive
+	if err := db.UpsertHost(ctx, s.db, host); err != nil {
+		return fmt.Errorf("upserting host status to db: %w", err)
+	}
+
+	return nil
 }
 
 func (s *KnotStream) ResubscribeAllHosts(ctx context.Context) error {
@@ -78,7 +88,7 @@
 	for _, host := range hosts {
 		l := s.logger.With("hostname", host.Hostname)
 		l.Info("re-subscribing to active host")
-		if err := s.slurper.Subscribe(ctx, host); err != nil {
+		if err := s.slurper.Subscribe(host); err != nil {
 			l.Warn("failed to re-subscribe to host", "err", err)
 		}
 		// sleep for a very short period, so we don't open tons of sockets at the same time
diff --git a/knotmirror/knotstream/slurper.go b/knotmirror/knotstream/slurper.go
--- a/knotmirror/knotstream/slurper.go
+++ b/knotmirror/knotstream/slurper.go
@@ -89,7 +89,7 @@ // s.logger.Info("finished persisting cursors", "count", len(cursors), "duration", time.Since(start).String(), "err", err)
 	return err
 }
 
-func (s *KnotSlurper) Subscribe(ctx context.Context, host models.Host) error {
+func (s *KnotSlurper) Subscribe(host models.Host) error {
 	s.subsLk.Lock()
 	defer s.subsLk.Unlock()
 
@@ -109,6 +109,10 @@ ),
 	}
 	s.subs[host.Hostname] = sub
 
+	// TODO: use service level context, not the top-most one.
+	// Using top-most context should be avoided to do graceful shutdown.
+	ctx := context.TODO()
+
 	sub.scheduler.Start(ctx)
 	go s.subscribeWithRedialer(ctx, host, sub)
 	return nil
@@ -120,6 +124,7 @@ defer func() {
 		s.subsLk.Lock()
 		defer s.subsLk.Unlock()
 
+		l.Info("unsubscribing knot")
 		delete(s.subs, host.Hostname)
 	}()
 

--
tangled.sh