Merge pull request #118 from benbjohnson/attr-sync

Add profile & bitmap attribute anti-entropy

attr.go

@@ -1,6 +1,8 @@
 package pilosa
 
 import (
+	"bytes"
+	"crypto/sha1"
 	"encoding/binary"
 	"fmt"
 	"sort"
@@ -12,6 +14,9 @@ import (
 	"github.com/umbel/pilosa/internal"
 )
 
+// AttrBlockSize is the size of attribute blocks for anti-entropy.
+const AttrBlockSize = 100
+
 // AttrStore represents a storage layer for attributes.
 type AttrStore struct {
 	mu   sync.Mutex
@@ -150,6 +155,69 @@ func (s *AttrStore) SetBulkAttrs(m map[uint64]map[string]interface{}) error {
 	return nil
 }
 
+// Blocks returns a list of all blocks in the store.
+func (s *AttrStore) Blocks() ([]AttrBlock, error) {
+	tx, err := s.db.Begin(false)
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback()
+
+	// Wrap cursor to segment by block.
+	cur := newBlockCursor(tx.Bucket([]byte("attrs")).Cursor(), AttrBlockSize)
+
+	// Iterate over each block.
+	var blocks []AttrBlock
+	for cur.nextBlock() {
+		block := AttrBlock{ID: cur.blockID()}
+
+		// Compute checksum of every key/value in block.
+		h := sha1.New()
+		for k, v := cur.next(); k != nil; k, v = cur.next() {
+			h.Write(k)
+			h.Write(v)
+		}
+		block.Checksum = h.Sum(nil)
+
+		// Append block.
+		blocks = append(blocks, block)
+	}
+
+	return blocks, nil
+}
+
+// BlockData returns all data for a single block.
+func (s *AttrStore) BlockData(i uint64) (map[uint64]map[string]interface{}, error) {
+	m := make(map[uint64]map[string]interface{})
+
+	// Start read-only transaction.
+	tx, err := s.db.Begin(false)
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback()
+
+	// Move to the start of the block.
+	min := u64tob(uint64(i) * AttrBlockSize)
+	max := u64tob(uint64(i+1) * AttrBlockSize)
+	cur := tx.Bucket([]byte("attrs")).Cursor()
+	for k, v := cur.Seek(min); k != nil; k, v = cur.Next() {
+		// Exit if we're past the end of the block.
+		if bytes.Compare(k, max) != -1 {
+			break
+		}
+
+		// Decode attribute map and associate with id.
+		var pb internal.AttrMap
+		if err := proto.Unmarshal(v, &pb); err != nil {
+			return nil, err
+		}
+		m[btou64(k)] = decodeAttrs(pb.GetAttrs())
+	}
+
+	return m, nil
+}
+
 // txAttrs returns a map of attributes for a bitmap.
 func txAttrs(tx *bolt.Tx, id uint64) (map[string]interface{}, error) {
 	v := tx.Bucket([]byte("attrs")).Get(u64tob(id))
@@ -284,3 +352,113 @@ func btou64(b []byte) uint64 { return binary.BigEndian.Uint64(b) }
 
 // emptyMap is a reusable map that contains no keys.
 var emptyMap = make(map[string]interface{})
+
+// AttrBlock represents a checksummed block of the attribute store.
+type AttrBlock struct {
+	ID       uint64 `json:"id"`
+	Checksum []byte `json:"checksum"`
+}
+
+// AttrBlocks represents a list of blocks.
+type AttrBlocks []AttrBlock
+
+// Diff returns a list of block ids that are different or are new in other.
+// Block lists must be in sorted order.
+func (a AttrBlocks) Diff(other []AttrBlock) []uint64 {
+	var ids []uint64
+	for {
+		// Read next block from each list.
+		var blk0, blk1 *AttrBlock
+		if len(a) > 0 {
+			blk0 = &a[0]
+		}
+		if len(other) > 0 {
+			blk1 = &other[0]
+		}
+
+		// Exit if "a" contains no more blocks.
+		if blk0 == nil {
+			return ids
+		}
+
+		// Add block ID if it's different or if it's only in "a".
+		if blk1 == nil || blk0.ID < blk1.ID {
+			ids = append(ids, blk0.ID)
+			a = a[1:]
+		} else if blk1.ID < blk0.ID {
+			other = other[1:]
+		} else {
+			if !bytes.Equal(blk0.Checksum, blk1.Checksum) {
+				ids = append(ids, blk0.ID)
+			}
+			a, other = a[1:], other[1:]
+		}
+	}
+}
+
+// blockCursor represents a cursor for iterating over blocks of a bolt bucket.
+type blockCursor struct {
+	cur  *bolt.Cursor
+	base uint64
+	n    uint64
+
+	buf struct {
+		key    []byte
+		value  []byte
+		filled bool
+	}
+}
+
+// newBlockCursor returns a new block cursor that wraps cur using n sized blocks.
+func newBlockCursor(c *bolt.Cursor, n int) blockCursor {
+	cur := blockCursor{
+		cur: c,
+		n:   uint64(n),
+	}
+	cur.buf.key, cur.buf.value = c.First()
+	cur.buf.filled = true
+	return cur
+}
+
+// blockID returns the current block ID. Only valid after call to nextBlock().
+func (cur *blockCursor) blockID() uint64 { return cur.base }
+
+// nextBlock moves the cursor to the next block.
+// Returns true if another block exists, otherwise returns false.
+func (cur *blockCursor) nextBlock() bool {
+	if cur.buf.key == nil {
+		return false
+	}
+
+	cur.base = binary.BigEndian.Uint64(cur.buf.key) / cur.n
+	return true
+}
+
+// next returns the next key/value within the block.
+// Returns nils at the end of the block.
+func (cur *blockCursor) next() (key, value []byte) {
+	// Use buffered value, if set.
+	if cur.buf.filled {
+		key, value = cur.buf.key, cur.buf.value
+		cur.buf.filled = false
+		return key, value
+	}
+
+	// Read next key.
+	key, value = cur.cur.Next()
+
+	// Fill buffer for EOF.
+	if key == nil {
+		cur.buf.key, cur.buf.value, cur.buf.filled = key, value, false
+		return nil, nil
+	}
+
+	// Parse key and buffer if outside of block.
+	id := binary.BigEndian.Uint64(key)
+	if id/cur.n > cur.base {
+		cur.buf.key, cur.buf.value, cur.buf.filled = key, value, true
+		return nil, nil
+	}
+
+	return key, value
+}

attr_test.go

@@ -70,6 +70,48 @@ func TestAttrStore_Attrs_Unset(t *testing.T) {
 	}
 }
 
+// Ensure attribute block checksums can be returned.
+func TestAttrStore_Blocks(t *testing.T) {
+	s := MustOpenAttrStore()
+	defer s.Close()
+
+	// Set attributes.
+	if err := s.SetAttrs(1, map[string]interface{}{"A": uint64(100)}); err != nil {
+		t.Fatal(err)
+	} else if err := s.SetAttrs(2, map[string]interface{}{"A": uint64(200)}); err != nil {
+		t.Fatal(err)
+	} else if err := s.SetAttrs(100, map[string]interface{}{"B": "VALUE"}); err != nil {
+		t.Fatal(err)
+	} else if err := s.SetAttrs(350, map[string]interface{}{"C": "FOO"}); err != nil {
+		t.Fatal(err)
+	}
+
+	// Retrieve blocks.
+	blks0, err := s.Blocks()
+	if err != nil {
+		t.Fatal(err)
+	} else if len(blks0) != 3 || blks0[0].ID != 0 || blks0[1].ID != 1 || blks0[2].ID != 3 {
+		t.Fatalf("unexpected blocks: %#v", blks0)
+	}
+
+	// Change second block.
+	if err := s.SetAttrs(100, map[string]interface{}{"X": 12}); err != nil {
+		t.Fatal(err)
+	}
+
+	// Ensure second block changed.
+	blks1, err := s.Blocks()
+	if err != nil {
+		t.Fatal(err)
+	} else if !reflect.DeepEqual(blks0[0], blks1[0]) {
+		t.Fatalf("block 0 mismatch: %#v != %#v", blks0[0], blks1[0])
+	} else if reflect.DeepEqual(blks0[1], blks1[1]) {
+		t.Fatalf("block 1 match: %#v ", blks0[0])
+	} else if !reflect.DeepEqual(blks0[2], blks1[2]) {
+		t.Fatalf("block 2 mismatch: %#v != %#v", blks0[2], blks1[2])
+	}
+}
+
 // AttrStore represents a test wrapper for pilosa.AttrStore.
 type AttrStore struct {
 	*pilosa.AttrStore

client.go

@@ -630,6 +630,80 @@ func (c *Client) BlockData(db, frame string, slice uint64, block int) ([]uint64,
 	return rsp.BitmapIDs, rsp.ProfileIDs, nil
 }
 
+// ProfileAttrDiff returns data from differing blocks on a remote host.
+func (c *Client) ProfileAttrDiff(db string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) {
+	u := url.URL{
+		Scheme:   "http",
+		Host:     c.host,
+		Path:     "/db/attr/diff",
+		RawQuery: url.Values{"db": {db}}.Encode(),
+	}
+
+	// Encode request.
+	buf, err := json.Marshal(postDBAttrDiffRequest{DB: db, Blocks: blks})
+	if err != nil {
+		return nil, err
+	}
+
+	// Send request.
+	resp, err := c.HTTPClient.Post(u.String(), "application/json", bytes.NewReader(buf))
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	// Return error if status is not OK.
+	switch resp.StatusCode {
+	case http.StatusOK: // ok
+	default:
+		return nil, fmt.Errorf("unexpected status: code=%d", resp.StatusCode)
+	}
+
+	// Decode response object.
+	var rsp postDBAttrDiffResponse
+	if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil {
+		return nil, err
+	}
+	return rsp.Attrs, nil
+}
+
+// BitmapAttrDiff returns data from differing blocks on a remote host.
+func (c *Client) BitmapAttrDiff(db, frame string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) {
+	u := url.URL{
+		Scheme:   "http",
+		Host:     c.host,
+		Path:     "/frame/attr/diff",
+		RawQuery: url.Values{"db": {db}, "frame": {frame}}.Encode(),
+	}
+
+	// Encode request.
+	buf, err := json.Marshal(postFrameAttrDiffRequest{DB: db, Frame: frame, Blocks: blks})
+	if err != nil {
+		return nil, err
+	}
+
+	// Send request.
+	resp, err := c.HTTPClient.Post(u.String(), "application/json", bytes.NewReader(buf))
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	// Return error if status is not OK.
+	switch resp.StatusCode {
+	case http.StatusOK: // ok
+	default:
+		return nil, fmt.Errorf("unexpected status: code=%d", resp.StatusCode)
+	}
+
+	// Decode response object.
+	var rsp postFrameAttrDiffResponse
+	if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil {
+		return nil, err
+	}
+	return rsp.Attrs, nil
+}
+
 // Bit represents the location of a single bit.
 type Bit struct {
 	BitmapID  uint64

cmd/pilosa/main.go

@@ -134,6 +134,9 @@ func (m *Main) Run(args ...string) error {
 	m.Server.Host = m.Config.Host
 	m.Server.Cluster = m.Config.PilosaCluster()
 
+	// Set configuration options.
+	m.Server.AntiEntropyInterval = time.Duration(m.Config.AntiEntropy.Interval)
+
 	// Initialize server.
 	if err := m.Server.Open(); err != nil {
 		return err

fragment.go

@@ -1254,6 +1254,7 @@ func (s *FragmentSyncer) SyncFragment() error {
 		//fmt.Println("no place to replicate", s.Fragment.DB(), s.Fragment.Frame(), s.Fragment.Slice())
 		return nil
 	}
+
 	// Create a set of blocks.
 	blockSets := make([][]FragmentBlock, 0, len(nodes))
 	for _, node := range nodes {

handler.go

@@ -106,6 +106,20 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		default:
 			http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
 		}
+	case "/db/attr/diff":
+		switch r.Method {
+		case "POST":
+			h.handlePostDBAttrDiff(w, r)
+		default:
+			http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		}
+	case "/frame/attr/diff":
+		switch r.Method {
+		case "POST":
+			h.handlePostFrameAttrDiff(w, r)
+		default:
+			http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		}
 	case "/fragment/nodes":
 		switch r.Method {
 		case "GET":
@@ -250,6 +264,119 @@ type sliceMaxResponse struct {
 	SliceMax uint64 `json:"SliceMax"`
 }
 
+// handlePostDBAttrDiff handles POST /db/attr/diff requests.
+func (h *Handler) handlePostDBAttrDiff(w http.ResponseWriter, r *http.Request) {
+	// Decode request.
+	var req postDBAttrDiffRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	// Retrieve database from index.
+	db, err := h.Index.CreateDBIfNotExists(req.DB)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	// Retrieve local blocks.
+	blks, err := db.ProfileAttrStore().Blocks()
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	// Read all attributes from all mismatched blocks.
+	attrs := make(map[uint64]map[string]interface{})
+	for _, blockID := range AttrBlocks(blks).Diff(req.Blocks) {
+		// Retrieve block data.
+		m, err := db.ProfileAttrStore().BlockData(blockID)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+
+		// Copy to database-wide struct.
+		for k, v := range m {
+			attrs[k] = v
+		}
+	}
+
+	// Encode response.
+	if err := json.NewEncoder(w).Encode(postDBAttrDiffResponse{
+		Attrs: attrs,
+	}); err != nil {
+		h.logger().Printf("response encoding error: %s", err)
+	}
+}
+
+type postDBAttrDiffRequest struct {
+	DB     string      `json:"db"`
+	Blocks []AttrBlock `json:"blocks"`
+}
+
+type postDBAttrDiffResponse struct {
+	Attrs map[uint64]map[string]interface{} `json:"attrs"`
+}
+
+// handlePostFrameAttrDiff handles POST /frame/attr/diff requests.
+func (h *Handler) handlePostFrameAttrDiff(w http.ResponseWriter, r *http.Request) {
+	// Decode request.
+	var req postFrameAttrDiffRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	// Retrieve database from index.
+	f, err := h.Index.CreateFrameIfNotExists(req.DB, req.Frame)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	// Retrieve local blocks.
+	blks, err := f.BitmapAttrStore().Blocks()
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	// Read all attributes from all mismatched blocks.
+	attrs := make(map[uint64]map[string]interface{})
+	for _, blockID := range AttrBlocks(blks).Diff(req.Blocks) {
+		// Retrieve block data.
+		m, err := f.BitmapAttrStore().BlockData(blockID)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+
+		// Copy to database-wide struct.
+		for k, v := range m {
+			attrs[k] = v
+		}
+	}
+
+	// Encode response.
+	if err := json.NewEncoder(w).Encode(postFrameAttrDiffResponse{
+		Attrs: attrs,
+	}); err != nil {
+		h.logger().Printf("response encoding error: %s", err)
+	}
+}
+
+type postFrameAttrDiffRequest struct {
+	DB     string      `json:"db"`
+	Frame  string      `json:"frame"`
+	Blocks []AttrBlock `json:"blocks"`
+}
+
+type postFrameAttrDiffResponse struct {
+	Attrs map[uint64]map[string]interface{} `json:"attrs"`
+}
+
 // readProfiles returns a list of profile objects by id.
 func (h *Handler) readProfiles(db *DB, ids []uint64) ([]*Profile, error) {
 	if db == nil {

handler_test.go

@@ -2,8 +2,10 @@ package pilosa_test
 
 import (
 	"bytes"
+	"encoding/json"
 	"errors"
 	"io"
+	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
@@ -414,6 +416,104 @@ func TestHandler_Query_ErrParse(t *testing.T) {
 	}
 }
 
+// Ensure the handler can return data in differing blocks for a database.
+func TestHandler_DB_AttrStore_Diff(t *testing.T) {
+	idx := MustOpenIndex()
+	defer idx.Close()
+
+	s := NewServer()
+	s.Handler.Index = idx.Index
+	defer s.Close()
+
+	// Set attributes on the database.
+	db, err := idx.CreateDBIfNotExists("d")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := db.ProfileAttrStore().SetAttrs(1, map[string]interface{}{"foo": 1, "bar": 2}); err != nil {
+		t.Fatal(err)
+	} else if err := db.ProfileAttrStore().SetAttrs(100, map[string]interface{}{"x": "y"}); err != nil {
+		t.Fatal(err)
+	} else if err := db.ProfileAttrStore().SetAttrs(200, map[string]interface{}{"snowman": "☃"}); err != nil {
+		t.Fatal(err)
+	}
+
+	// Retrieve block checksums.
+	blks, err := db.ProfileAttrStore().Blocks()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Remove block #0 and alter block 2's checksum.
+	blks = blks[1:]
+	blks[1].Checksum = []byte("MISMATCHED_CHECKSUM")
+
+	// Send block checksums to determine diff.
+	resp, err := http.Post(
+		s.URL+"/db/attr/diff?db=d",
+		"application/json",
+		strings.NewReader(`{"db":"d", "blocks":`+string(MustMarshalJSON(blks))+`}`),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+
+	// Read and validate body.
+	if body := string(MustReadAll(resp.Body)); body != `{"attrs":{"1":{"bar":2,"foo":1},"200":{"snowman":"☃"}}}`+"\n" {
+		t.Fatalf("unexpected body: %s", body)
+	}
+}
+
+// Ensure the handler can return data in differing blocks for a frame.
+func TestHandler_Frame_AttrStore_Diff(t *testing.T) {
+	idx := MustOpenIndex()
+	defer idx.Close()
+
+	s := NewServer()
+	s.Handler.Index = idx.Index
+	defer s.Close()
+
+	// Set attributes on the database.
+	f, err := idx.CreateFrameIfNotExists("d", "f")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := f.BitmapAttrStore().SetAttrs(1, map[string]interface{}{"foo": 1, "bar": 2}); err != nil {
+		t.Fatal(err)
+	} else if err := f.BitmapAttrStore().SetAttrs(100, map[string]interface{}{"x": "y"}); err != nil {
+		t.Fatal(err)
+	} else if err := f.BitmapAttrStore().SetAttrs(200, map[string]interface{}{"snowman": "☃"}); err != nil {
+		t.Fatal(err)
+	}
+
+	// Retrieve block checksums.
+	blks, err := f.BitmapAttrStore().Blocks()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Remove block #0 and alter block 2's checksum.
+	blks = blks[1:]
+	blks[1].Checksum = []byte("MISMATCHED_CHECKSUM")
+
+	// Send block checksums to determine diff.
+	resp, err := http.Post(
+		s.URL+"/frame/attr/diff?db=d",
+		"application/json",
+		strings.NewReader(`{"db":"d", "frame":"f", "blocks":`+string(MustMarshalJSON(blks))+`}`),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+
+	// Read and validate body.
+	if body := string(MustReadAll(resp.Body)); body != `{"attrs":{"1":{"bar":2,"foo":1},"200":{"snowman":"☃"}}}`+"\n" {
+		t.Fatalf("unexpected body: %s", body)
+	}
+}
+
 // Ensure the handler can backup a fragment and then restore it.
 func TestHandler_Fragment_BackupRestore(t *testing.T) {
 	idx := MustOpenIndex()
@@ -570,3 +670,21 @@ func MustNewHTTPRequest(method, urlStr string, body io.Reader) *http.Request {
 	}
 	return req
 }
+
+// MustMarshalJSON marshals v to JSON. Panic on error.
+func MustMarshalJSON(v interface{}) []byte {
+	buf, err := json.Marshal(v)
+	if err != nil {
+		panic(err)
+	}
+	return buf
+}
+
+// MustReadAll reads a reader into a buffer and returns it. Panic on error.
+func MustReadAll(r io.Reader) []byte {
+	buf, err := ioutil.ReadAll(r)
+	if err != nil {
+		panic(err)
+	}
+	return buf
+}

index.go

@@ -275,13 +275,43 @@ type IndexSyncer struct {
 	Closing <-chan struct{}
 }
 
+// Returns true if the syncer has been marked to close.
+func (s *IndexSyncer) IsClosing() bool {
+	select {
+	case <-s.Closing:
+		return true
+	default:
+		return false
+	}
+}
+
 // SyncIndex compares the index on host with the local index and resolves differences.
 func (s *IndexSyncer) SyncIndex() error {
 	sliceN := s.Index.SliceN()
 
 	// Iterate over schema in sorted order.
 	for _, di := range s.Index.Schema() {
+		// Verify syncer has not closed.
+		if s.IsClosing() {
+			return nil
+		}
+
+		// Sync database profile attributes.
+		if err := s.syncDatabase(di.Name); err != nil {
+			return fmt.Errorf("db sync error: db=%s, err=%s", di.Name, err)
+		}
+
 		for _, fi := range di.Frames {
+			// Verify syncer has not closed.
+			if s.IsClosing() {
+				return nil
+			}
+
+			// Sync frame bitmap attributes.
+			if err := s.syncFrame(di.Name, fi.Name); err != nil {
+				return fmt.Errorf("frame sync error: db=%s, frame=%s, err=%s", di.Name, fi.Name, err)
+			}
+
 			for slice := uint64(0); slice <= sliceN; slice++ {
 				// Ignore slices that this host doesn't own.
 				if !s.Cluster.OwnsFragment(s.Host, di.Name, slice) {
@@ -289,15 +319,13 @@ func (s *IndexSyncer) SyncIndex() error {
 				}
 
 				// Verify syncer has not closed.
-				select {
-				case <-s.Closing:
+				if s.IsClosing() {
 					return nil
-				default:
 				}
 
 				// Sync fragment if own it.
 				if err := s.syncFragment(di.Name, fi.Name, slice); err != nil {
-					return fmt.Errorf("sync error: db=%s, frame=%s, slice=%d, err=%s", di.Name, fi.Name, slice, err)
+					return fmt.Errorf("fragment sync error: db=%s, frame=%s, slice=%d, err=%s", di.Name, fi.Name, slice, err)
 				}
 			}
 		}
@@ -306,7 +334,97 @@ func (s *IndexSyncer) SyncIndex() error {
 	return nil
 }
 
-// syncFragment synchronizes
+// syncDatabase synchronizes database attributes with the rest of the cluster.
+func (s *IndexSyncer) syncDatabase(db string) error {
+	// Retrieve database reference.
+	d := s.Index.DB(db)
+	if d == nil {
+		return nil
+	}
+
+	// Read block checksums.
+	blks, err := d.ProfileAttrStore().Blocks()
+	if err != nil {
+		return err
+	}
+
+	// Sync with every other host.
+	for _, node := range Nodes(s.Cluster.Nodes).FilterHost(s.Host) {
+		client, err := NewClient(node.Host)
+		if err != nil {
+			return err
+		}
+
+		// Retrieve attributes from differing blocks.
+		// Skip update and recomputation if no attributes have changed.
+		m, err := client.ProfileAttrDiff(db, blks)
+		if err != nil {
+			return err
+		} else if len(m) == 0 {
+			continue
+		}
+
+		// Update local copy.
+		if err := d.ProfileAttrStore().SetBulkAttrs(m); err != nil {
+			return err
+		}
+
+		// Recompute blocks.
+		blks, err = d.ProfileAttrStore().Blocks()
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// syncFrame synchronizes frame attributes with the rest of the cluster.
+func (s *IndexSyncer) syncFrame(db, name string) error {
+	// Retrieve database reference.
+	f := s.Index.Frame(db, name)
+	if f == nil {
+		return nil
+	}
+
+	// Read block checksums.
+	blks, err := f.BitmapAttrStore().Blocks()
+	if err != nil {
+		return err
+	}
+
+	// Sync with every other host.
+	for _, node := range Nodes(s.Cluster.Nodes).FilterHost(s.Host) {
+		client, err := NewClient(node.Host)
+		if err != nil {
+			return err
+		}
+
+		// Retrieve attributes from differing blocks.
+		// Skip update and recomputation if no attributes have changed.
+		m, err := client.BitmapAttrDiff(db, name, blks)
+		if err != nil {
+			return err
+		} else if len(m) == 0 {
+			continue
+		}
+
+		// Update local copy.
+		if err := f.BitmapAttrStore().SetBulkAttrs(m); err != nil {
+			return err
+		}
+
+		// Recompute blocks.
+		blks, err = f.BitmapAttrStore().Blocks()
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// syncFragment synchronizes a fragment with the rest of the cluster.
 func (s *IndexSyncer) syncFragment(db, frame string, slice uint64) error {
 	// Ensure fragment exists locally.
 	f, err := s.Index.CreateFragmentIfNotExists(db, frame, slice)

server.go

@@ -147,10 +147,10 @@ func (s *Server) Addr() net.Addr {
 func (s *Server) logger() *log.Logger { return log.New(s.LogOutput, "", log.LstdFlags) }
 
 func (s *Server) monitorAntiEntropy() {
-	ticker := time.NewTicker(time.Duration(s.AntiEntropyInterval))
+	ticker := time.NewTicker(s.AntiEntropyInterval)
 	defer ticker.Stop()
 
-	s.logger().Printf("index sync monitor initializing")
+	s.logger().Printf("index sync monitor initializing (%s interval)", s.AntiEntropyInterval)
 
 	for {
 		// Wait for tick or a close.