diff --git a/api.go b/api.go index e0c4236f9..6149c6aee 100644 --- a/api.go +++ b/api.go @@ -328,11 +328,17 @@ func (api *API) CreateField(ctx context.Context, indexName string, fieldName str } // Create field. - field, err := index.CreateFieldAndBroadcast(cfm) + field, err := index.CreateField(fieldName, opts...) if err != nil { return nil, errors.Wrap(err, "creating field") } + // Send the create field message to all nodes. We do this *outside* the + // CreateField logic so we're not blocking on it. + if err := api.holder.sendOrSpool(cfm); err != nil { + return nil, errors.Wrap(err, "sending CreateField message") + } + api.holder.Stats.CountWithCustomTags(MetricCreateField, 1, 1.0, []string{fmt.Sprintf("index:%s", indexName)}) return field, nil } diff --git a/api_test.go b/api_test.go index 82ce1af4f..3b4836fab 100644 --- a/api_test.go +++ b/api_test.go @@ -29,6 +29,8 @@ import ( "github.com/molecula/featurebase/v3/shardwidth" "github.com/molecula/featurebase/v3/test" . "github.com/molecula/featurebase/v3/vprint" // nolint:staticcheck + + "golang.org/x/sync/errgroup" ) func TestAPI_Import(t *testing.T) { @@ -1403,6 +1405,42 @@ func TestVariousApiTranslateCalls(t *testing.T) { } } +func TestAPI_CreateField(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + c := test.MustRunCluster(t, 3) + defer c.Close() + + nodes := make([]*test.Command, 3) + for i := range nodes { + nodes[i] = c.GetNode(i) + } + + if _, err := nodes[0].API.CreateIndex(ctx, "i", pilosa.IndexOptions{}); err != nil { + t.Fatal(err) + } + eg, ctx := errgroup.WithContext(context.Background()) + for _, n := range nodes { + node := n + eg.Go(func() error { + for i := 0; i < 10; i++ { + _, err := node.API.CreateField(ctx, "i", fmt.Sprintf("f%d", i)) + if err != nil && !errors.Is(err, pilosa.ErrFieldExists) { + return err + } + } + return nil + }) + } + err := eg.Wait() + if err != nil { + if errors.Is(err, pilosa.ErrFieldExists) { + t.Fatalf("conflict error: %v", err) + } + t.Fatalf("unexpected error: %T %v", err, err) + } +} + func TestAPI_RBFDebugInfo(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() diff --git a/index.go b/index.go index 66932cd32..12e0aee68 100644 --- a/index.go +++ b/index.go @@ -503,12 +503,21 @@ func (i *Index) CreateField(name string, opts ...FieldOption) (*Field, error) { return nil, errors.Wrap(err, "validating name") } - i.mu.Lock() - defer i.mu.Unlock() + // Grab lock, check for field existing, release lock. We don't want + // to stay holding the lock, but we might care about the ErrFieldExists + // part of this. + err = func() error { + i.mu.Lock() + defer i.mu.Unlock() - // Ensure field doesn't already exist. - if i.fields[name] != nil { - return nil, newConflictError(ErrFieldExists) + // Ensure field doesn't already exist. + if i.fields[name] != nil { + return newConflictError(ErrFieldExists) + } + return nil + }() + if err != nil { + return nil, err } // Apply and validate functional options. @@ -524,37 +533,26 @@ func (i *Index) CreateField(name string, opts ...FieldOption) (*Field, error) { Meta: fo, } - // Create the field in etcd as the system of record. + // Create the field in etcd as the system of record. We do this without + // the lock held because it can take an arbitrary amount of time... if err := i.persistField(context.Background(), cfm); err != nil { return nil, errors.Wrap(err, "persisting field") } - return i.createField(cfm, false) -} - -// CreateFieldAndBroadcast creates a field locally, then broadcasts the -// creation to other nodes so they can create locally as well. An error is -// returned if the field already exists. -func (i *Index) CreateFieldAndBroadcast(cfm *CreateFieldMessage) (*Field, error) { - err := ValidateName(cfm.Field) - if err != nil { - return nil, errors.Wrap(err, "validating name") - } - + // This is identical to the previous check, because we could get super + // unlucky and have the persist-field thing happen, and somehow the field + // gets created, before we get to run again, and the specific nature of + // the error can matter to the backend. i.mu.Lock() defer i.mu.Unlock() // Ensure field doesn't already exist. - if i.fields[cfm.Field] != nil { + if i.fields[name] != nil { return nil, newConflictError(ErrFieldExists) } - // Create the field in etcd as the system of record. - if err := i.persistField(context.Background(), cfm); err != nil { - return nil, errors.Wrap(err, "persisting field") - } - - return i.createField(cfm, true) + // Actually do the internal bookkeeping. + return i.createField(cfm) } // CreateFieldIfNotExists creates a field with the given options if it doesn't exist. @@ -594,7 +592,7 @@ func (i *Index) CreateFieldIfNotExists(name string, opts ...FieldOption) (*Field return nil, errors.Wrap(err, "persisting field") } - return i.createField(cfm, false) + return i.createField(cfm) } // CreateFieldIfNotExistsWithOptions is a method which I created because I @@ -632,7 +630,7 @@ func (i *Index) CreateFieldIfNotExistsWithOptions(name string, opt *FieldOptions return nil, errors.Wrap(err, "persisting field") } - return i.createField(cfm, false) + return i.createField(cfm) } // persistField stores the field information in etcd. @@ -667,14 +665,14 @@ func (i *Index) createFieldIfNotExists(cfm *CreateFieldMessage) (*Field, error) return f, nil } - return i.createField(cfm, false) + return i.createField(cfm) } -// createField, in addition to creating a new Field, calls Field.Open which -// potentially aquires a lock on Index. So until/unless we refactor the -// Index.createField() function call path, we cannot call Index.createField -// while holding an Index lock. -func (i *Index) createField(cfm *CreateFieldMessage, broadcast bool) (*Field, error) { +// createField does the internal field creation logic, creating the in-memory +// data structure, and kicking translation sync if appropriate. It does not +// notify other nodes; that's done from the API's initial CreateField call +// now. +func (i *Index) createField(cfm *CreateFieldMessage) (*Field, error) { opt := cfm.Meta if opt == nil { opt = &FieldOptions{} @@ -711,13 +709,6 @@ func (i *Index) createField(cfm *CreateFieldMessage, broadcast bool) (*Field, er // enable Txf to find the index in field_test.go TestField_SetValue f.idx = i - if broadcast { - // Send the create field message to all nodes. - if err := i.holder.sendOrSpool(cfm); err != nil { - return nil, errors.Wrap(err, "sending CreateField message") - } - } - // Kick off the field's translation sync process. if err := i.translationSyncer.Reset(); err != nil { return nil, errors.Wrap(err, "resetting translation syncer") diff --git a/pilosa.go b/pilosa.go index 9cf4f715f..218e37491 100644 --- a/pilosa.go +++ b/pilosa.go @@ -111,6 +111,12 @@ func newConflictError(err error) ConflictError { return ConflictError{err} } +// Unwrap makes it so that a ConflictError wrapping ErrFieldExists gets a +// true from errors.Is(ErrFieldExists). +func (c ConflictError) Unwrap() error { + return c.error +} + // NotFoundError wraps an error value to signify that a resource was not found // such that in an HTTP scenario, http.StatusNotFound would be returned. type NotFoundError error