code: public void addWatch(DatabaseWatch watch, 
                         TableKraken table, 
                         byte[] key,
                         Result<Cancel> result)
    {
      WatchTable watchTable = getWatchTable(table);
      
      watchTable.addWatchLocal(watch, key);
      
      // bfs/112b
      // XXX: table.isKeyLocalCopy vs table.isKeyLocalPrimary
      // XXX: using copy because a copy server reads its own data directly
      // XXX: and the timing between a remote watch and the put can result in
      // XXX: the wrong version
      // XXX: (old) notify requires isKeyLocalOwner, otherwise the notify isn't propagated
  
      if (! table.isKeyLocalCopy(key)) {
        HashKey hashKey = HashKey.create(key);
        Set<HashKey> keys = _remoteKeys.get(table);
        
        if (keys == null) {
          keys = new HashSet<>();
          _remoteKeys.put(table, keys);
        }
          
        if (! keys.contains(hashKey)) {
          table.addRemoteWatch(key);
          
          keys.add(hashKey);
        }
        
        
        // getWatchTable(table).addWatch(watch, key);
      }
      
      WatchHandle handle = new WatchHandle(table, key, watch);
      
      result.ok(_serviceRef.pin(handle).as(Cancel.class));
    }
docs: /* XXX:
  private void onPodUpdate(PodBartender pod)
  {
  for (Map.Entry<TableKraken,Set<HashKey>> remoteKeys : _remoteKeys.entrySet()) {
  TableKraken table = remoteKeys.getKey();
  
  if (! table.getTablePod().getPodName().equals(pod.name())) {
  continue;
  }
  
  for (HashKey hashKey : remoteKeys.getValue()) {
  table.addRemoteWatch(hashKey.getHash());
  }
  }
  }