As I mentioned Tail/Feather is a weekend project to test out how stuff works for real. After creating the highly available distributed key/value store, we are now in need of actually building a client API for it.

Externally, that API is going to look like this:

public class TailFeatherClient : IDisposable
{
    public TailFeatherClient(params Uri[] nodes);

    public Task Set(string key, JToken value);

    public Task<JToken> Get(string key);

    public Task Remove(string key);

    public void Dispose();
}

If this wasn’t a weekend project, I would add batch support, but that isn’t important for our purposes right now. The API itself is pretty stupid, which is great, but what about the actual behavior?

We want it to be able to handle dynamic cluster changes, and we need it to be smart about it. A lot of that is shared among all operations, so the next layer of the API is:

public Task Set(string key, JToken value)
{
    return ContactServer(client => client.GetAsync(string.Format("tailfeather/key-val/set?key={0}&val={1}", 
        Uri.EscapeDataString(key), Uri.EscapeDataString(value.ToString(Formatting.None)))));
}

public async Task<JToken> Get(string key)
{
    var reply = await ContactServer(client => client.GetAsync(string.Format("tailfeather/key-val/del?key={0}",
        Uri.EscapeDataString(key))));
    var result = JObject.Load(new JsonTextReader(new StreamReader(await reply.Content.ReadAsStreamAsync())));

    if (result.Value<bool>("Missing"))
        return null;

    return result["Value"];
}

public Task Remove(string key)
{
    return ContactServer(client => client.GetAsync(string.Format("tailfeather/key-val/del?key={0}",
        Uri.EscapeDataString(key))));
}

The actual behavior is in ContactServer:

private readonly ConcurrentDictionary<Uri, HttpClient> _cache = new ConcurrentDictionary<Uri, HttpClient>();
private Task<TailFeatherTopology> _topologyTask;

public TailFeatherClient(params Uri[] nodes)
{
    _topologyTask = FindLatestTopology(nodes);
}

private HttpClient GetHttpClient(Uri node)
{
    return _cache.GetOrAdd(node, uri => new HttpClient { BaseAddress = uri });
}

private async Task<TailFeatherTopology> FindLatestTopology(IEnumerable<Uri> nodes)
{
    var tasks = nodes.Select(node => GetHttpClient(node).GetAsync("tailfeather/admin/flock")).ToArray();

    await Task.WhenAny(tasks);
    var topologies = new List<TailFeatherTopology>();
    foreach (var task in tasks)
    {
        var message = task.Result;
        if (message.IsSuccessStatusCode == false)
            continue;

        topologies.Add(new JsonSerializer().Deserialize<TailFeatherTopology>(
            new JsonTextReader(new StreamReader(await message.Content.ReadAsStreamAsync()))));
    }

    return topologies.OrderByDescending(x => x.CommitIndex).FirstOrDefault();
}

private async Task<HttpResponseMessage> ContactServer(Func<HttpClient, Task<HttpResponseMessage>> operation, int retries = 3)
{
    if (retries < 0)
        throw new InvalidOperationException("Cluster is not reachable, or no leader was selected. Out of retries, aborting.");

    var topology = (await _topologyTask ?? new TailFeatherTopology());
        
    var leader = topology.AllVotingNodes.FirstOrDefault(x => x.Name == topology.CurrentLeader);
    if (leader == null)
    {
        _topologyTask = FindLatestTopology(topology.AllVotingNodes.Select(x => x.Uri));
        return await ContactServer(operation, retries - 1);
    }

    // now we have a leader, we need to try calling it...
    var httpResponseMessage = await operation(GetHttpClient(leader.Uri));
    if (httpResponseMessage.IsSuccessStatusCode == false)
    {
        // we were sent to a different server, let try that...
        if (httpResponseMessage.StatusCode == HttpStatusCode.Redirect)
        {
            var redirectUri = httpResponseMessage.Headers.Location;
            httpResponseMessage = await operation(GetHttpClient(redirectUri));
            if (httpResponseMessage.IsSuccessStatusCode)
            {
                // we successfully contacted the redirected server, this is probably the leader, let us ask it for the topology,
                // it will be there for next time we access it
                _topologyTask = FindLatestTopology(new[] { redirectUri }.Union(topology.AllVotingNodes.Select(x => x.Uri)));

                return httpResponseMessage;
            }
        }

        // we couldn't get to the server, and we didn't get redirected, we'll check in the cluster in general
        _topologyTask = FindLatestTopology(topology.AllVotingNodes.Select(x => x.Uri));
        return await ContactServer(operation, retries - 1);
    }

    // happy path, we are done
    return httpResponseMessage;
}

There is quite a bit going on here. But the basic idea is simple. Starting from the initial list of nodes we have, contact all of them and find the topology with the highest commit index. That means that it is the freshest, so more likely to be the current one. From the topology, we take the leader, and send all queries to the leader.

If there is any sort of errors, we’ll contact all other servers to find who we are supposed to be using now. If we can’t find it after three tries, we give us and we let the caller sort it out, probably by retrying once the cluster is in a steady state again.

Now, this is really nice, but it is falling into the heading of weekend code. That is means that this is quite far from what I would call production code. What is missing?

• Caching the topology locally in a persistent manner so we can restart when the known servers are down from last known good topology.
• Proper error handling, and in particular, error reporting, to make sure that we can understand what is actually is going on.
• Features such as allowing reads from non leaders, testing, etc.

But overall, I’m quite happy with this.