Co-authored-by: Travis Turner <github@calfrope.com>

cluster.go

@@ -1443,7 +1443,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error {
 		return errors.Wrap(err, "merging cluster status")
 	}
 
-	c.logger.Printf("done MergeClusterStatus, start goroutine")
+	c.logger.Printf("done MergeClusterStatus, start goroutine (%s)", c.Node.ID)
 
 	// The actual resizing runs in a goroutine because we don't want to block
 	// the distribution of other ResizeInstructions to the rest of the cluster.
@@ -1482,6 +1482,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error {
 						c.logger.Printf("local field not found: %s/%s", is.Name, fs.Name)
 						continue
 					}
+					fmt.Printf("field: %+v  (%s)\n", fs.AvailableShards.Slice(), c.Node.ID)
 					if err := f.AddRemoteAvailableShards(fs.AvailableShards); err != nil {
 						return errors.Wrap(err, "adding remote available shards")
 					}

field.go

@@ -375,7 +375,6 @@ func newField(path, index, name string, opts FieldOption) (*Field, error) {
 
 		OpenTranslateStore: OpenInMemTranslateStore,
 	}
-	f.options.ContainsShard = f.containsShard //used for notification optimization
 
 	return f, nil
 }
@@ -417,6 +416,8 @@ func (f *Field) AvailableShards() *roaring.Bitmap {
 func (f *Field) containsShard(shard uint64) bool {
 	f.mu.RLock()
 	defer f.mu.RUnlock()
+	fmt.Println("CONTAINS SHARD:", shard, f.Name())
+	fmt.Println(f.remoteAvailableShards)
 	return f.remoteAvailableShards.Contains(shard)
 }
 
@@ -608,7 +609,7 @@ func (f *Field) Open() error {
 	f.logger.Debugf("successfully opened field index/field: %s/%s", f.index, f.name)
 	return nil
 }
-func saveIt(fieldPath string, availableShardBytes []byte) {
+func blockingWriteAvailableShards(fieldPath string, availableShardBytes []byte) {
 	path := filepath.Join(fieldPath, ".available.shards")
 	// Create a temporary file to save to.
 	tempPath := path + tempExt
@@ -624,12 +625,12 @@ func saveIt(fieldPath string, availableShardBytes []byte) {
 	}
 
 }
-func saveItAsync(fieldPath string, availableShardBytes []byte, done chan bool) {
+func nonBlockingWriteAvailableShards(fieldPath string, availableShardBytes []byte, done chan bool) {
 	if len(availableShardBytes) == 0 {
 		return
 	}
 	go func() {
-		saveIt(fieldPath, availableShardBytes)
+		blockingWriteAvailableShards(fieldPath, availableShardBytes)
 		done <- true
 	}()
 }
@@ -649,18 +650,18 @@ func (f *Field) writeAvailableShards() {
 			if len(data) > 0 {
 				if !writing {
 					writing = true
-					saveItAsync(f.path, data, tracker)
+					nonBlockingWriteAvailableShards(f.path, data, tracker)
 					data = nil
 				}
 			}
 		case <-tracker:
 			writing = false
 		case <-f.doneChan:
-			if writing {
+			if writing { //wait to writing is complete
 				<-tracker
 			}
 			if len(data) > 0 {
-				saveIt(f.path, data)
+				blockingWriteAvailableShards(f.path, data)
 			}
 			alive = false
 		}
@@ -1186,6 +1187,7 @@ func (f *Field) newView(path, name string) *view {
 	view.rowAttrStore = f.rowAttrStore
 	view.stats = f.Stats
 	view.broadcaster = f.broadcaster
+	view.shardPresent = f.containsShard
 	if f.snapshotQueue != nil {
 		view.snapshotQueue = f.snapshotQueue
 	}
@@ -2005,16 +2007,12 @@ type FieldOptions struct {
 	Type           string      `json:"type,omitempty"`
 	TimeQuantum    TimeQuantum `json:"timeQuantum,omitempty"`
 	ForeignIndex   string      `json:"foreignIndex"`
-	ContainsShard  func(uint64) bool
 }
 
 // newFieldOptions returns a new instance of FieldOptions
 // with applied and validated functional options.
 func newFieldOptions(opts ...FieldOption) (*FieldOptions, error) {
 	fo := FieldOptions{}
-	fo.ContainsShard = func(uint64) bool {
-		return false
-	}
 	for _, opt := range opts {
 		err := opt(&fo)
 		if err != nil {
@@ -2043,8 +2041,6 @@ func applyDefaultOptions(o *FieldOptions) *FieldOptions {
 		o.CacheType = DefaultCacheType
 		o.CacheSize = DefaultCacheSize
 	}
-	o.ContainsShard = func(uint64) bool { return false } //used for shardnotify optimization
-
 	return o
 }
 

holder.go

@@ -1185,7 +1185,7 @@ func (s *holderSyncer) readFieldTranslateReader(rd TranslateEntryReader) {
 		// Find appropriate store.
 		f := s.Holder.Field(entry.Index, entry.Field)
 		if f == nil {
-			s.Holder.Logger.Printf("field not found: %q/%q", entry.Index, entry.Field)
+			s.Holder.Logger.Printf("field not found: %s/%s", entry.Index, entry.Field)
 			return
 		}
 

server/cluster_test.go

@@ -138,7 +138,7 @@ func TestClusterResize_EmptyNodes(t *testing.T) {
 }
 
 // Ensure that adding a node correctly resizes the cluster.
-func TestClusterResize_AddNode(t *testing.T) {
+func TestXClusterResize_AddNode(t *testing.T) {
 	t.Run("NoData", func(t *testing.T) {
 		clus := test.MustRunCluster(t, 2)
 		defer clus.Close()
@@ -206,7 +206,6 @@ func TestClusterResize_AddNode(t *testing.T) {
 			`); err != nil {
 			t.Fatal(err)
 		}
-
 		// exp is the expected result for the Row queries that follow.
 		exp := `{"results":[{"attrs":{},"columns":[1,1300000]}]}` + "\n"
 
@@ -388,6 +387,8 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) {
 			t.Fatal(err)
 		} else if res != exp {
 			t.Fatalf("unexpected result: %s", res)
+		} else {
+			fmt.Println(res)
 		}
 
 		// Configure node1

view.go

@@ -80,7 +80,7 @@ func newView(path, index, field, name string, fieldOptions FieldOptions) *view {
 		broadcaster:  NopBroadcaster,
 		stats:        stats.NopStatsClient,
 		logger:       logger.NopLogger,
-		shardPresent: fieldOptions.ContainsShard,
+		shardPresent: func(uint64) bool { return false },
 	}
 }
 
@@ -283,10 +283,11 @@ func (v *view) CreateFragmentIfNotExists(shard uint64) (*fragment, error) {
 }
 
 func (v *view) notifyIfNew(shard uint64) {
+	fmt.Println("Present", shard)
 	if v.shardPresent(shard) {
 		return
 	}
-
+	fmt.Println("BROADCAST", shard)
 	broadcastChan := make(chan struct{})
 
 	go func() {

view_internal_test.go

@@ -30,9 +30,8 @@ func mustOpenView(index, field, name string) *view {
 	}
 
 	fo := FieldOptions{
-		CacheType:     DefaultCacheType,
-		CacheSize:     DefaultCacheSize,
-		ContainsShard: func(uint64) bool { return false },
+		CacheType: DefaultCacheType,
+		CacheSize: DefaultCacheSize,
 	}
 
 	v := newView(path, index, field, name, fo)