Don't block on Query call for in-process plugins

pkg/plugins/clients/in_process.go

@@ -22,7 +22,9 @@ import (
 
 	v1 "github.com/webmeshproj/api/v1"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/emptypb"
 
 	"github.com/webmeshproj/node/pkg/context"
@@ -49,10 +51,16 @@ func (p *inProcessPlugin) Query(ctx context.Context, opts ...grpc.CallOption) (v
 	schan := make(chan *v1.PluginSQLQuery)
 	rchan := make(chan *v1.PluginSQLQueryResult)
 	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
 	srv := &inProcessQueryServer{ctx, schan, rchan}
 	cli := &inProcessQueryClient{ctx, cancel, schan, rchan}
-	return cli, p.server.Query(srv)
+	go func() {
+		defer cancel()
+		err := p.server.Query(srv)
+		if err != nil && err != io.EOF && status.Code(err) != codes.Unimplemented {
+			context.LoggerFrom(ctx).Error("error in plugin query", "err", err)
+		}
+	}()
+	return cli, nil
 }
 
 type inProcessQueryClient struct {

pkg/plugins/ipam/plugin.go

@@ -80,6 +80,10 @@ func (p *Plugin) Configure(ctx context.Context, req *v1.PluginConfiguration) (*e
 	return &emptypb.Empty{}, nil
 }
 
+func (p *Plugin) Query(srv v1.Plugin_QueryServer) error {
+	return nil
+}
+
 func (p *Plugin) Close(ctx context.Context, req *emptypb.Empty) (*emptypb.Empty, error) {
 	return &emptypb.Empty{}, p.data.Close()
 }

pkg/plugins/manager.go

@@ -254,12 +254,14 @@ func (m *manager) handleQueryClient(plugin string, client clients.PluginClient,
 		query, err := queries.Recv()
 		if err != nil {
 			if err == io.EOF {
+				m.log.Debug("query stream closed cleanly", "plugin", plugin)
 				return
 			}
 			// TODO: restart the stream?
 			m.log.Error("receive query", "plugin", plugin, "error", err)
 			return
 		}
+		m.log.Debug("handling plugin query", "plugin", plugin, "query", query.GetQuery())
 		var result v1.PluginSQLQueryResult
 		result.Id = query.GetId()
 		res, err := raftlogs.QueryWithQuerier(queries.Context(), m.db, query.GetQuery())