diff --git a/services/data_service/cmd/stream_tap/main.go b/services/data_service/cmd/stream_tap/main.go
index 67e8cbe..d870359 100644
--- a/services/data_service/cmd/stream_tap/main.go
+++ b/services/data_service/cmd/stream_tap/main.go
@@ -92,7 +92,7 @@ func main() {
 	flag.Parse()
 
 	if len(ids) == 0 {
-		fmt.Fprintln(os.Stderr, "provide at least one --id (provider:subject or canonical key)")
+		_, _ = fmt.Fprintln(os.Stderr, "provide at least one --id (provider:subject or canonical key)")
 		os.Exit(2)
 	}
 
@@ -104,16 +104,24 @@ func main() {
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
 	)
 	if err != nil {
-		fmt.Fprintf(os.Stderr, "new control client: %v\n", err)
+		_, _ = fmt.Fprintf(os.Stderr, "new control client: %v\n", err)
 		os.Exit(1)
 	}
-	defer ccCtl.Close()
+	defer func(ccCtl *grpc.ClientConn) {
+		err := ccCtl.Close()
+		if err != nil {
+			_, _ = fmt.Fprintf(os.Stderr, "close control client: %v\n", err)
+			os.Exit(1)
+		} else {
+			fmt.Println("closed control client")
+		}
+	}(ccCtl)
 	ccCtl.Connect()
 
 	ctlConnCtx, cancelCtlConn := context.WithTimeout(ctx, timeout)
 	if err := waitReady(ctlConnCtx, ccCtl); err != nil {
 		cancelCtlConn()
-		fmt.Fprintf(os.Stderr, "connect control: %v\n", err)
+		_, _ = fmt.Fprintf(os.Stderr, "connect control: %v\n", err)
 		os.Exit(1)
 	}
 	cancelCtlConn()
@@ -124,7 +132,7 @@ func main() {
 	startResp, err := ctl.StartStream(ctxStart, &pb.StartStreamRequest{})
 	cancelStart()
 	if err != nil {
-		fmt.Fprintf(os.Stderr, "StartStream: %v\n", err)
+		_, _ = fmt.Fprintf(os.Stderr, "StartClientStream: %v\n", err)
 		os.Exit(1)
 	}
 	streamUUID := startResp.GetStreamUuid()
@@ -134,7 +142,7 @@ func main() {
 	for _, s := range ids {
 		key, err := toIdentifierKey(s)
 		if err != nil {
-			fmt.Fprintf(os.Stderr, "bad --id: %v\n", err)
+			_, _ = fmt.Fprintf(os.Stderr, "bad --id: %v\n", err)
 			os.Exit(2)
 		}
 		pbIDs = append(pbIDs, &pb.Identifier{Key: key})
@@ -147,7 +155,7 @@ func main() {
 	})
 	cancelCfg()
 	if err != nil {
-		fmt.Fprintf(os.Stderr, "ConfigureStream: %v\n", err)
+		_, _ = fmt.Fprintf(os.Stderr, "ConfigureClientStream: %v\n", err)
 		os.Exit(1)
 	}
 	fmt.Printf("configured %d identifiers\n", len(pbIDs))
@@ -157,16 +165,24 @@ func main() {
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
 	)
 	if err != nil {
-		fmt.Fprintf(os.Stderr, "new streaming client: %v\n", err)
+		_, _ = fmt.Fprintf(os.Stderr, "new streaming client: %v\n", err)
 		os.Exit(1)
 	}
-	defer ccStr.Close()
+	defer func(ccStr *grpc.ClientConn) {
+		err := ccStr.Close()
+		if err != nil {
+			_, _ = fmt.Fprintf(os.Stderr, "close streaming client: %v\n", err)
+			os.Exit(1)
+		} else {
+			fmt.Println("closed streaming client")
+		}
+	}(ccStr)
 	ccStr.Connect()
 
 	strConnCtx, cancelStrConn := context.WithTimeout(ctx, timeout)
 	if err := waitReady(strConnCtx, ccStr); err != nil {
 		cancelStrConn()
-		fmt.Fprintf(os.Stderr, "connect streaming: %v\n", err)
+		_, _ = fmt.Fprintf(os.Stderr, "connect streaming: %v\n", err)
 		os.Exit(1)
 	}
 	cancelStrConn()
@@ -178,7 +194,7 @@ func main() {
 
 	stream, err := str.ConnectStream(streamCtx, &pb.ConnectStreamRequest{StreamUuid: streamUUID})
 	if err != nil {
-		fmt.Fprintf(os.Stderr, "ConnectStream: %v\n", err)
+		_, _ = fmt.Fprintf(os.Stderr, "ConnectClientStream: %v\n", err)
 		os.Exit(1)
 	}
 	fmt.Println("connected; streaming… (Ctrl-C to quit)")
@@ -194,7 +210,7 @@ func main() {
 				if ctx.Err() != nil {
 					return
 				}
-				fmt.Fprintf(os.Stderr, "recv: %v\n", err)
+				_, _ = fmt.Fprintf(os.Stderr, "recv: %v\n", err)
 				os.Exit(1)
 			}
 			id := msg.GetIdentifier()
diff --git a/services/data_service/internal/manager/manager.go b/services/data_service/internal/manager/manager.go
index e0a6326..ef1fc8f 100644
--- a/services/data_service/internal/manager/manager.go
+++ b/services/data_service/internal/manager/manager.go
@@ -39,7 +39,7 @@ func NewManager(router *router.Router) *Manager {
 	}
 }
 
-func (m *Manager) StartStream() (uuid.UUID, error) {
+func (m *Manager) StartClientStream() (uuid.UUID, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
@@ -50,7 +50,7 @@ func (m *Manager) StartStream() (uuid.UUID, error) {
 		OutChannel:  nil,
 		Timer: time.AfterFunc(1*time.Minute, func() {
 			fmt.Printf("stream %s expired due to inactivity\n", streamID)
-			err := m.StopStream(streamID)
+			err := m.StopClientStream(streamID)
 			if err != nil {
 				fmt.Printf("failed to stop stream after timeout: %v\n", err)
 			}
@@ -60,7 +60,7 @@ func (m *Manager) StartStream() (uuid.UUID, error) {
 	return streamID, nil
 }
 
-func (m *Manager) ConfigureStream(streamID uuid.UUID, newIds []domain.Identifier) error {
+func (m *Manager) ConfigureClientStream(streamID uuid.UUID, newIds []domain.Identifier) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
@@ -150,8 +150,8 @@ func (m *Manager) ConfigureStream(streamID uuid.UUID, newIds []domain.Identifier
 	return nil
 }
 
-func (m *Manager) StopStream(streamID uuid.UUID) error {
-	m.DisconnectStream(streamID)
+func (m *Manager) StopClientStream(streamID uuid.UUID) error {
+	m.DisconnectClientStream(streamID)
 
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -186,7 +186,7 @@ func (m *Manager) StopStream(streamID uuid.UUID) error {
 	return nil
 }
 
-func (m *Manager) ConnectStream(streamID uuid.UUID) (<-chan domain.Message, error) {
+func (m *Manager) ConnectClientStream(streamID uuid.UUID) (<-chan domain.Message, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
@@ -214,7 +214,7 @@ func (m *Manager) ConnectStream(streamID uuid.UUID) (<-chan domain.Message, erro
 	return ch, nil
 }
 
-func (m *Manager) DisconnectStream(streamID uuid.UUID) {
+func (m *Manager) DisconnectClientStream(streamID uuid.UUID) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
@@ -232,7 +232,7 @@ func (m *Manager) DisconnectStream(streamID uuid.UUID) {
 
 	stream.Timer = time.AfterFunc(1*time.Minute, func() {
 		fmt.Printf("stream %s expired due to inactivity\n", streamID)
-		err := m.StopStream(streamID)
+		err := m.StopClientStream(streamID)
 		if err != nil {
 			fmt.Printf("failed to stop stream after disconnect: %v\n", err)
 		}
diff --git a/services/data_service/internal/server/gprc_control_server.go b/services/data_service/internal/server/gprc_control_server.go
index c0aee6b..52f180f 100644
--- a/services/data_service/internal/server/gprc_control_server.go
+++ b/services/data_service/internal/server/gprc_control_server.go
@@ -22,7 +22,7 @@ func NewGRPCControlServer(m *manager.Manager) *GRPCControlServer {
 }
 
 func (s *GRPCControlServer) StartStream(_ context.Context, _ *pb.StartStreamRequest) (*pb.StartStreamResponse, error) {
-	streamID, err := s.manager.StartStream()
+	streamID, err := s.manager.StartClientStream()
 	if err != nil {
 		return nil, fmt.Errorf("failed to start stream: %w", err)
 	}
@@ -44,7 +44,7 @@ func (s *GRPCControlServer) ConfigureStream(_ context.Context, req *pb.Configure
 		ids = append(ids, id)
 	}
 
-	if err := s.manager.ConfigureStream(streamID, ids); err != nil {
+	if err := s.manager.ConfigureClientStream(streamID, ids); err != nil {
 		return nil, status.Errorf(codes.InvalidArgument, "configure failed: %v", err)
 	}
 	return &pb.ConfigureStreamResponse{}, nil
@@ -55,7 +55,7 @@ func (s *GRPCControlServer) StopStream(_ context.Context, req *pb.StopStreamRequ
 	if err != nil {
 		return nil, status.Errorf(codes.InvalidArgument, "invalid stream_uuid %q: %v", req.StreamUuid, err)
 	}
-	if err := s.manager.StopStream(streamID); err != nil {
+	if err := s.manager.StopClientStream(streamID); err != nil {
 		return nil, status.Errorf(codes.Internal, "failed to stop stream: %v", err)
 	}
 	return &pb.StopStreamResponse{}, nil
diff --git a/services/data_service/internal/server/grpc_streaming_server.go b/services/data_service/internal/server/grpc_streaming_server.go
index 120caa3..c0768f6 100644
--- a/services/data_service/internal/server/grpc_streaming_server.go
+++ b/services/data_service/internal/server/grpc_streaming_server.go
@@ -23,7 +23,7 @@ func (s *GRPCStreamingServer) ConnectStream(req *pb.ConnectStreamRequest, stream
 		return fmt.Errorf("invalid UUID: %w", err)
 	}
 
-	ch, err := s.manager.ConnectStream(streamUUID)
+	ch, err := s.manager.ConnectClientStream(streamUUID)
 	if err != nil {
 		return fmt.Errorf("failed to connect: %w", err)
 	}
@@ -31,7 +31,7 @@ func (s *GRPCStreamingServer) ConnectStream(req *pb.ConnectStreamRequest, stream
 	for {
 		select {
 		case <-stream.Context().Done():
-			s.manager.DisconnectStream(streamUUID)
+			s.manager.DisconnectClientStream(streamUUID)
 			return nil
 		case msg, ok := <-ch:
 			if !ok {
diff --git a/services/data_service/internal/server/socket_streaming_server.go b/services/data_service/internal/server/socket_streaming_server.go
index 423f508..5401027 100644
--- a/services/data_service/internal/server/socket_streaming_server.go
+++ b/services/data_service/internal/server/socket_streaming_server.go
@@ -66,12 +66,12 @@ func (s *SocketStreamingServer) handleConnection(conn net.Conn) {
 		return
 	}
 
-	outCh, err := s.manager.ConnectStream(streamUUID)
+	outCh, err := s.manager.ConnectClientStream(streamUUID)
 	if err != nil {
 		_, _ = fmt.Fprintf(conn, "Failed to connect to stream: %v\n", err)
 		return
 	}
-	defer s.manager.DisconnectStream(streamUUID)
+	defer s.manager.DisconnectClientStream(streamUUID)
 
 	writer := bufio.NewWriterSize(conn, 256*1024)
 	defer func(w *bufio.Writer) {
diff --git a/services/data_service/internal/worker/stateful.go b/services/data_service/internal/worker/stateful.go
new file mode 100644
index 0000000..4df0094
--- /dev/null
+++ b/services/data_service/internal/worker/stateful.go
@@ -0,0 +1 @@
+package worker
diff --git a/services/data_service/internal/worker/stateless.go b/services/data_service/internal/worker/stateless.go
new file mode 100644
index 0000000..4df0094
--- /dev/null
+++ b/services/data_service/internal/worker/stateless.go
@@ -0,0 +1 @@
+package worker