cannot cancel watch

[fofewyoung]

Describe the bug
i try to cancel a watch with below code, but failed.

To Reproduce

using Etcdserverpb;
using Google.Protobuf;
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Text;

namespace dotnet_etcd
{
    class TestCancelWatch
    {
        dotnet_etcd.EtcdClient myConsulClient;

        const long WID = 4445;

        public void Start()
        {
            myConsulClient = new dotnet_etcd.EtcdClient("http://127.0.0.1:2379");
            WatchKey("/zzz");
        }

        public void Stop()
        {
            UnWatch();
        }

        void WatchKey(string keyName)
        {
            WatchRequest req = new WatchRequest {
                CreateRequest = new WatchCreateRequest
                {
                    Key = ByteString.CopyFromUtf8(keyName),
                    WatchId = WID,
                }
            };

            myConsulClient.Watch(req, (Etcdserverpb.WatchResponse respon) => {
                Debug.Assert(respon.WatchId == WID);
                Console.WriteLine("watch event!!!");
            });
        }

        // unwatch
        public void UnWatch()
        {
            WatchRequest req = new WatchRequest
            {
                CancelRequest = new WatchCancelRequest
                {
                    WatchId = WID
                }
            };

            myConsulClient.Watch(req, (Etcdserverpb.WatchResponse respon) => {

                // never be executed!!!! bug?
                Console.WriteLine($"cancel:{respon.Canceled}: {respon.CancelReason}");
            });
        }
    }


    class Program
    {
        static void Main(string[] args)
        {
            var watch = new TestCancelWatch();

            while (true)
            {
                var str = Console.ReadLine();
                if (str == "s")
                {
                    watch.Start();
                }
                else if (str == "u")
                {
                    watch.Stop();
                }
            }
        }
    }
}

Expected behavior
input 's'
input 'u'
console print "cancel: ..."

Screenshots

Additional context
dotnet-etcd: 3.0.0
etcd:3.4.3

[fofewyoung]

Hi, May I check if you read my questions please?
Maybe I know the reason, but i have no idea to fix it simply,
in dotnet-etcd, each watch operation has its own gRPC stream,
but cancel watch operation should use the streaming of watch.
cancel operation works after i hack the code to let all watch* share a single streaming.
Could you please help me deal with it?

[shubhamranjan]

Hi, May I check if you read my questions please?
Maybe I know the reason, but i have no idea to fix it simply,
in dotnet-etcd, each watch operation has its own gRPC stream,
but cancel watch operation should use the streaming of watch.
cancel operation works after i hack the code to let all watch* share a single streaming.
Could you please help me deal with it?

Hi, I have read your question. Will get back once I reproduce the issue and if a problem exists, will fix it.

Do let me know about your findings and how did you modify the code to get it working.

[shubhamranjan]

but cancel watch operation should use the streaming of watch.

Strange. What would be the use of watchId then. Need to dig a bit more in this.

[fofewyoung]

hi, this is my hack code :
it is not thread safe, and not graceful

It should be noted that Annotated code:
// await watcher.RequestStream.CompleteAsync();

 public partial class EtcdClient : IDisposable
 {
        // the share streaming
        AsyncDuplexStreamingCall<WatchRequest, WatchResponse> myWatchStreaming = null;
        AsyncDuplexStreamingCall<WatchRequest, WatchResponse> GetWatchStreaming()
        {
            if (myWatchStreaming == null)
                myWatchStreaming = _balancer.GetConnection().watchClient.Watch(null);

            return myWatchStreaming;
        }
        
         // unwatch operation
        public async void UnWatch(long wid)
        {
            WatchRequest req = new WatchRequest
            {
                CancelRequest = new WatchCancelRequest
                {
                    WatchId = wid
                }
            };

            await GetWatchStreaming().RequestStream.WriteAsync(req);
        }

        #region Watch Key
        /// <summary>
        /// Watches a key according to the specified watch request and
        /// passes the watch response to the method provided.
        /// </summary>
        /// <param name="request">Watch Request containing key to be watched</param>
        /// <param name="method">Method to which watch response should be passed on</param>
        public async void Watch(WatchRequest request, Action<WatchResponse> method, Metadata headers = null)
        {
            bool success = false;
            int retryCount = 0;
            while (!success)
            {
                try
                {
                    var watcher = GetWatchStreaming();
                    {
                        Task watcherTask = Task.Run(async () =>
                        {
                            while (await watcher.ResponseStream.MoveNext())
                            {
                                WatchResponse update = watcher.ResponseStream.Current;
                                if (update.Canceled)
                                    break;

                                method(update);
                            }
                        });

                        await watcher.RequestStream.WriteAsync(request);
                        //await watcher.RequestStream.CompleteAsync();    // prevent  streaming to be closed
                        await watcherTask;
                    }
                    success = true;
                }
                catch (RpcException ex) when (ex.StatusCode == StatusCode.Unavailable)
                {
                    retryCount++;
                    if (retryCount >= _balancer._numNodes)
                    {
                        throw ex;
                    }
                }
            }

        }
 }

[shubhamranjan]

Thanks. I will look into this. A bit of a busy week, may not be able to come up with a solution soon enough.

[shubhamranjan]

Update: I was able to replicate the issue. Still working on the solution.

[shubhamranjan]

Lets take an example where we have an etcd cluster and an application using this client. Suppose I have 3 instances of my application behind a load balancer and my application connects to etcd using this client. Now if I a make an api call to my application via loadbalancer for cancel watch request, I would never be able to gurantee for the request to land on the same server from where the watch was started making it difficult to cancel the watch. Will have to look into it why the watch ID is not being honored by etcd.

I will take this up ahead with etcd team.

[yoricksmeets]

The watchId exists to manage different watches on the same stream to the server and is only scoped to that stream. See this discussion here.

What I assume is that if you close the response stream (or maybe both request and response stream but in this client the request stream is always closed after sending the initial requests) the watch is also cancelled, but I could not confirm this assumption yet.

[shubhamranjan]

So probably, we will have to maintain request streams on our end. Will try to re-design the watch implementation here to persist request streams. The discussion also states that each stream may share the same watch ID which becomes a bit difficult to manage on client side.

[yoricksmeets]

@shubhamranjan I'm working on a watch manager to use in our distributed locking client. I can share the code when I have progressed a bit more. You could get some inspiration from it or we could adjust it to be included in this library if you think it fits the scope of this library.

The basic idea is as follows (WM = Watch Manager, WMS = Watch Manager Subscription):

• The WM is created from a etcdclient instance and the WM will use this etcdclient instance to get AsyncDuplexStreamingCall for its request and response stream
• The WM can issue WMS that are tracked by the WM
• The WM will issue unique watchIds for all WMS issued by this WM
• The WM creates a task that reads the response stream in a seperate task and dispatches the responses to the correct WMS based on the watchId
• The WM will respond to any exceptions in the response stream reading task, if the stream fails the WM will resubscribe all WMS that it tracks on a newly created channel
• WMS will implement IDisposable: disposing the WMS will cancel the subscription
• WM will implement IDisposable: disposing the WM will close the grpc channel and dispose all WMS it was tracking
• The etcd client will track all issued WMs and will dispose them in its own Dispose function

I'm not sure if a WMS needs to track versions of the keys it receives so when the WM needs to create a new stream and resubscribe it knows what the start version is that it needs to receive, making sure the WMS only receives every version once even if we need to create a new channel.

A developer now can get a WatchManger that can issue multiple Subscriptions using a single duplex gprc stream. A Subscription will manage its own lifetime and by disposing the Subscription object the watch will be cancelled on the etcd server.

[shubhamranjan]

The WM will issue unique watchIds for all WMS issued by this WM.

I would like to see how you plan to achieve this in case of same etcd watch ID(s) issued in different streams. Because in case of connection exceptions, I believe the stream would be recreated and etcd issued watch ID(s) won't be valid anymore for the new stream.

Looks fine to me if its included in this library. Would make the watch client better.