Merge pull request #1681 from seebs/core850

mutex sanity-check endpoints to allow for checking possible mutex corruption

api.go

@@ -2653,6 +2653,277 @@ func (api *API) RestoreShard(ctx context.Context, indexName string, shard uint64
 	return nil
 }
 
+func (api *API) mutexCheckThisNode(ctx context.Context, qcx *Qcx, indexName string, fieldName string) (map[uint64]map[uint64][]uint64, error) {
+	index := api.holder.Index(indexName)
+	if index == nil {
+		return nil, newNotFoundError(ErrIndexNotFound, indexName)
+	}
+	field := index.Field(fieldName)
+	if field == nil {
+		return nil, newNotFoundError(ErrFieldNotFound, fieldName)
+	}
+	return field.MutexCheck(ctx, qcx)
+}
+
+// mergeIDLists merges a list of numeric IDs into another list, removing
+// duplicates.
+func mergeIDLists(dst []uint64, src []uint64) []uint64 {
+	dst = append(dst, src...)
+	sort.Slice(dst, func(i, j int) bool {
+		return dst[i] < dst[j]
+	})
+	// dedup.
+	n := 0
+	prev := dst[0]
+	for i := 0; i < len(dst); i++ {
+		if dst[i] != prev {
+			dst[n] = dst[i]
+			n++
+		}
+		prev = dst[i]
+	}
+	return dst[:n]
+}
+
+// mergeKeyLists merges a list of string IDs into another list, removing
+// duplicates.
+func mergeKeyLists(dst []string, src []string) []string {
+	dst = append(dst, src...)
+	sort.Slice(dst, func(i, j int) bool {
+		return dst[i] < dst[j]
+	})
+	// dedup.
+	n := 0
+	prev := dst[0]
+	for i := 0; i < len(dst); i++ {
+		if dst[i] != prev {
+			dst[n] = dst[i]
+			n++
+		}
+		prev = dst[i]
+	}
+	return dst[:n]
+}
+
+// MutexCheckNode checks for collisions in a given mutex field. The response is
+// a map[shard]map[column]values, not translated.
+func (api *API) MutexCheckNode(ctx context.Context, qcx *Qcx, indexName string, fieldName string) (map[uint64]map[uint64][]uint64, error) {
+	if err := api.validate(apiMutexCheck); err != nil {
+		return nil, errors.Wrap(err, "validating api method")
+	}
+	return api.mutexCheckThisNode(ctx, qcx, indexName, fieldName)
+}
+
+// MutexCheck checks a named field for mutex violations, returning a
+// map of record IDs to values for records that have multiple values in the
+// field. The return will be one of:
+//     map[uint64][]uint64 // unkeyed index, unkeyed field
+//     map[uint64][]string // unkeyed index, keyed field
+//     map[string][]uint64 // keyed index, unkeyed field
+//     map[string][]string // keyed index, keyed field
+func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fieldName string) (result interface{}, err error) {
+	if err = api.validate(apiMutexCheck); err != nil {
+		return nil, errors.Wrap(err, "validating api method")
+	}
+	index, err := api.Index(ctx, indexName)
+	if err != nil {
+		return nil, err
+	}
+	field, err := api.Field(ctx, indexName, fieldName)
+	if err != nil {
+		return nil, err
+	}
+	if field.Type() != FieldTypeMutex {
+		return nil, errors.New("can only check mutex state for mutex fields")
+	}
+	// request data from other nodes as well
+	snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
+	eg, _ := errgroup.WithContext(ctx)
+	myID := api.NodeID()
+	results := make([]map[uint64]map[uint64][]uint64, len(snap.Nodes))
+	for i, node := range snap.Nodes {
+		i := i // loop variable shadowing is a war crime
+		if node.ID != myID {
+			node := node // loop variable shadowing again
+			eg.Go(func() (err error) {
+				results[i], err = api.server.defaultClient.MutexCheck(ctx, &node.URI, indexName, fieldName)
+				return err
+			})
+		} else {
+			eg.Go(func() (err error) {
+				results[i], err = api.mutexCheckThisNode(ctx, qcx, indexName, fieldName)
+				return err
+			})
+		}
+	}
+	err = eg.Wait()
+	if err != nil {
+		return nil, err
+	}
+	// We now have a series of maps from shards to maps of record IDs to
+	// values. But wait! Either the field, or the index, might be using keys,
+	// and want those translated. So we have to translate those. We'll create
+	// some tables.
+	useIndexKeys := index.Keys()
+	useFieldKeys := field.Keys()
+	var indexKeys = map[uint64]string{}
+	var fieldKeys = map[uint64]string{}
+	var indexIDs []uint64
+	var fieldIDs []uint64
+	// We'll use the string "untranslated" as our default value and overwrite
+	// it with translations. We do check for missing translation values in
+	// our returns, but just in case, you know?
+	untranslated := "untranslated"
+	// We don't know which of four map types we want to be working with,
+	// but what we can do is make a function which works with that map type
+	// given the raw integer values, and is a closure with an already-created
+	// map which has already been stashed in `result`. Because maps are
+	// reference-y, this should actually work.
+	var process func(uint64, []uint64)
+	if useIndexKeys || useFieldKeys {
+		for _, nodeResults := range results {
+			for _, shardResults := range nodeResults {
+				for record, values := range shardResults {
+					if useIndexKeys {
+						if _, ok := indexKeys[record]; !ok {
+							indexKeys[record] = untranslated
+							indexIDs = append(indexIDs, record)
+						}
+					}
+					if useFieldKeys {
+						for _, value := range values {
+							if _, ok := fieldKeys[value]; !ok {
+								fieldKeys[value] = untranslated
+								fieldIDs = append(fieldIDs, value)
+							}
+						}
+					}
+				}
+			}
+		}
+		untranslatedKeys := 0
+		// Obtain translation tables for the keys.
+		if useIndexKeys {
+			indexKeyList, err := api.cluster.translateIndexIDs(ctx, indexName, indexIDs)
+			if err != nil {
+				return nil, errors.Wrap(err, "translating index keys")
+			}
+			if len(indexKeyList) != len(indexIDs) {
+				return nil, fmt.Errorf("translating %d record IDs, got %d keys", len(indexIDs), len(indexKeyList))
+			}
+			for i := range indexIDs {
+				if indexKeyList[i] != "" {
+					indexKeys[indexIDs[i]] = indexKeyList[i]
+				} else {
+					untranslatedKeys++
+				}
+			}
+		}
+		if useFieldKeys {
+			fieldKeyList, err := api.cluster.translateFieldListIDs(field, fieldIDs)
+			if err != nil {
+				return nil, errors.Wrap(err, "translating index keys")
+			}
+			if len(fieldKeyList) != len(fieldIDs) {
+				return nil, fmt.Errorf("translating %d IDs, got %d keys", len(indexIDs), len(fieldKeyList))
+			}
+			for i := range fieldIDs {
+				if fieldKeyList[i] != "" {
+					fieldKeys[fieldIDs[i]] = fieldKeyList[i]
+				} else {
+					untranslatedKeys++
+				}
+			}
+		}
+		if untranslatedKeys > 0 {
+			api.server.logger.Warnf("translating mutex check results: %d key(s) untranslated", untranslatedKeys)
+		}
+	}
+
+	// define the process functions. separated from above code just to make
+	// it easier to follow/compare them.
+	if useIndexKeys {
+		if useFieldKeys {
+			outMap := make(map[string][]string)
+			var valueKeys []string
+			result = outMap
+			process = func(recordID uint64, valueIDs []uint64) {
+				valueKeys = valueKeys[:0]
+				for _, id := range valueIDs {
+					valueKeys = append(valueKeys, fieldKeys[id])
+				}
+				record := indexKeys[recordID]
+				if existing, ok := outMap[record]; ok {
+					outMap[record] = mergeKeyLists(existing, valueKeys)
+				} else {
+					// The append is so we can reuse this buffer safely,
+					// which matters if there's replication, because many
+					// cases won't need to copy the buffer, they'll just
+					// copy individual things from it.
+					outMap[record] = append([]string{}, valueKeys...)
+				}
+			}
+		} else {
+			outMap := make(map[string][]uint64)
+			result = outMap
+			process = func(recordID uint64, values []uint64) {
+				record := indexKeys[recordID]
+				if existing, ok := outMap[record]; ok {
+					outMap[record] = mergeIDLists(existing, values)
+				} else {
+					outMap[record] = values
+				}
+			}
+		}
+	} else {
+		if useFieldKeys {
+			outMap := make(map[uint64][]string)
+			var valueKeys []string
+			result = outMap
+			process = func(record uint64, valueIDs []uint64) {
+				valueKeys = valueKeys[:0]
+				for _, id := range valueIDs {
+					valueKeys = append(valueKeys, fieldKeys[id])
+				}
+				if existing, ok := outMap[record]; ok {
+					outMap[record] = mergeKeyLists(existing, valueKeys)
+				} else {
+					// The append is so we can reuse this buffer safely,
+					// which matters if there's replication, because many
+					// cases won't need to copy the buffer, they'll just
+					// copy individual things from it.
+					outMap[record] = append([]string{}, valueKeys...)
+				}
+			}
+		} else {
+			outMap := make(map[uint64][]uint64)
+			result = outMap
+			process = func(record uint64, values []uint64) {
+				if existing, ok := outMap[record]; ok {
+					outMap[record] = mergeIDLists(existing, values)
+				} else {
+					outMap[record] = values
+				}
+			}
+		}
+	}
+
+	for _, nodeResults := range results {
+		if len(nodeResults) == 0 {
+			continue
+		}
+		for _, v := range nodeResults {
+			if len(v) == 0 {
+				continue
+			}
+			for record, values := range v {
+				process(record, values)
+			}
+		}
+	}
+	return result, nil
+}
+
 type serverInfo struct {
 	ShardWidth       uint64 `json:"shardWidth"`
 	ReplicaN         int    `json:"replicaN"`
@@ -2713,6 +2984,7 @@ const (
 	apiIDReset
 	apiPartitionNodes
 	apiIngestOperations
+	apiMutexCheck
 )
 
 var methodsCommon = map[apiMethod]struct{}{
@@ -2781,4 +3053,5 @@ var methodsNormal = map[apiMethod]struct{}{
 	apiIDReset:              {},
 	apiPartitionNodes:       {},
 	apiIngestOperations:     {},
+	apiMutexCheck:           {},
 }

api_test.go

@@ -30,6 +30,7 @@ import (
 	"github.com/molecula/featurebase/v2/boltdb"
 	"github.com/molecula/featurebase/v2/http"
 	"github.com/molecula/featurebase/v2/server"
+	"github.com/molecula/featurebase/v2/shardwidth"
 	"github.com/molecula/featurebase/v2/test"
 	. "github.com/molecula/featurebase/v2/vprint" // nolint:staticcheck
 )
@@ -844,3 +845,316 @@ func TestAPI_IDAlloc(t *testing.T) {
 		}
 	})
 }
+
+type mutexCheckIndex struct {
+	index     *pilosa.Index
+	indexName string
+	createdAt int64
+	fields    map[bool]mutexCheckField
+}
+
+type mutexCheckField struct {
+	fieldName string
+	field     *pilosa.Field
+	createdAt int64
+}
+
+func TestAPI_MutexCheck(t *testing.T) {
+	c := test.MustRunCluster(t, 3)
+	defer c.Close()
+
+	m0 := c.GetNode(0)
+	nodesByID := make(map[string]*test.Command, 3)
+	qcxsByID := make(map[string]*pilosa.Qcx, 3)
+	for i := 0; i < 3; i++ {
+		node := c.GetNode(i)
+		id := node.API.NodeID()
+		nodesByID[id] = node
+	}
+
+	indexes := make(map[bool]mutexCheckIndex)
+
+	ctx := context.Background()
+	for _, keyedIndex := range []bool{false, true} {
+		indexName := fmt.Sprintf("i%t", keyedIndex)
+		index, err := m0.API.CreateIndex(ctx, indexName, pilosa.IndexOptions{Keys: keyedIndex, TrackExistence: true})
+		if err != nil {
+			t.Fatalf("creating index: %v", err)
+		}
+		if index.CreatedAt() == 0 {
+			t.Fatal("index createdAt is empty")
+		}
+		indexData := mutexCheckIndex{indexName: indexName, index: index, fields: make(map[bool]mutexCheckField), createdAt: index.CreatedAt()}
+		for _, keyedField := range []bool{false, true} {
+			fieldName := fmt.Sprintf("f%t", keyedField)
+			var field *pilosa.Field
+			if keyedField {
+				field, err = m0.API.CreateField(ctx, indexName, fieldName, pilosa.OptFieldTypeMutex(pilosa.CacheTypeNone, 0), pilosa.OptFieldKeys())
+			} else {
+				field, err = m0.API.CreateField(ctx, indexName, fieldName, pilosa.OptFieldTypeMutex(pilosa.CacheTypeNone, 0))
+			}
+			if err != nil {
+				t.Fatalf("creating field: %v", err)
+			}
+			if field.CreatedAt() == 0 {
+				t.Fatal("field createdAt is empty")
+			}
+			indexData.fields[keyedField] = mutexCheckField{fieldName: fieldName, field: field, createdAt: field.CreatedAt()}
+		}
+		indexes[keyedIndex] = indexData
+	}
+
+	rowIDs := []uint64{0, 1, 2, 3}
+	colIDs := []uint64{0, 1, 2, 3}
+	rowKeysBase := []string{"v0", "v1", "v2", "v3"}
+	colKeysBase := []string{"c0", "c1", "c2", "c3"}
+
+	const nShards = 10
+
+	// now, try the same thing for each combination of keyed/unkeyed. we
+	// share code between keyed/unkeyed fields, but for indexes, the logic
+	// is fundamentally different because we can't know shards in advance.
+	indexData := indexes[false]
+	for keyedField, fieldData := range indexData.fields {
+		t.Run(fmt.Sprintf("%s-%s", indexData.indexName, fieldData.fieldName), func(t *testing.T) {
+			for id, node := range nodesByID {
+				qcxsByID[id] = node.API.Txf().NewQcx()
+			}
+			for shard := uint64(0); shard < nShards; shard++ {
+				// restore row/col ID values which can get altered by imports
+				for i := range rowIDs {
+					rowIDs[i] = uint64(i)
+					colIDs[i] = (shard << shardwidth.Exponent) + uint64(i) + (shard % 4)
+				}
+				req := &pilosa.ImportRequest{
+					Index:          indexData.indexName,
+					IndexCreatedAt: indexData.createdAt,
+					Field:          fieldData.fieldName,
+					FieldCreatedAt: fieldData.createdAt,
+					Shard:          shard,
+					ColumnIDs:      colIDs,
+				}
+				if keyedField {
+					req.RowKeys = rowKeysBase
+				} else {
+					req.RowIDs = rowIDs
+				}
+				nodesForShard, err := m0.API.ShardNodes(ctx, indexData.indexName, shard)
+				if err != nil {
+					t.Fatalf("obtaining shard list: %v", err)
+				}
+				if len(nodesForShard) < 1 {
+					t.Fatalf("no nodes for shard %d", shard)
+				}
+				node := nodesByID[nodesForShard[0].ID]
+				if err := node.API.Import(ctx, qcxsByID[nodesForShard[0].ID], req); err != nil {
+					t.Fatalf("importing data: %v", err)
+				}
+			}
+			// and then we break the mutex and close the Qcxs
+			for id, node := range nodesByID {
+				field, err := node.API.Field(ctx, indexData.indexName, fieldData.fieldName)
+				if err != nil {
+					t.Fatalf("requesting field %s from node %s: %v", fieldData.fieldName, id, err)
+				}
+				pilosa.CorruptAMutex(t, field, qcxsByID[id])
+				err = qcxsByID[id].Finish()
+				if err != nil {
+					t.Fatalf("closing out transaction on node %s: %v", id, err)
+				}
+			}
+			qcx := m0.API.Txf().NewQcx()
+			defer qcx.Abort()
+
+			results, err := m0.API.MutexCheck(ctx, qcx, indexData.indexName, fieldData.fieldName)
+			if err != nil {
+				t.Fatalf("checking mutexes: %v", err)
+			}
+			// first two shards of each group of 4 should have a collision in
+			// position 1
+			expected := map[uint64]bool{
+				(0 << shardwidth.Exponent) + 1: true,
+				(1 << shardwidth.Exponent) + 1: true,
+				(4 << shardwidth.Exponent) + 1: true,
+				(5 << shardwidth.Exponent) + 1: true,
+				(8 << shardwidth.Exponent) + 1: true,
+				(9 << shardwidth.Exponent) + 1: true,
+			}
+			if keyedField {
+				mapped, ok := results.(map[uint64][]string)
+				if !ok {
+					t.Fatalf("expected map[uint64][]string, got %T", results)
+				}
+				seen := 0
+				for k, v := range mapped {
+					seen++
+					if !expected[k] {
+						t.Fatalf("expected all collisions to be 1 shards (s %% 4 in [0,1]), got %d", k)
+					}
+					if len(v) != 2 {
+						t.Fatalf("expected exactly two collisions")
+					}
+				}
+				if seen != len(expected) {
+					t.Fatalf("expected exactly %d records to have collisions", len(expected))
+				}
+			} else {
+				mapped, ok := results.(map[uint64][]uint64)
+				if !ok {
+					t.Fatalf("expected map[uint64][]uint64, got %T", results)
+				}
+				seen := 0
+				for k, v := range mapped {
+					seen++
+					if !expected[k] {
+						t.Fatalf("expected all collisions to be 1 shards (s %% 4 in [0,1]), got %d", k)
+					}
+					if len(v) != 2 {
+						t.Fatalf("expected exactly two collisions")
+					}
+				}
+				if seen != len(expected) {
+					t.Fatalf("expected exactly %d records to have collisions", len(expected))
+				}
+			}
+		})
+	}
+	indexData = indexes[true]
+	for keyedField, fieldData := range indexData.fields {
+		t.Run(fmt.Sprintf("%s-%s", indexData.indexName, fieldData.fieldName), func(t *testing.T) {
+			for id, node := range nodesByID {
+				qcxsByID[id] = node.API.Txf().NewQcx()
+			}
+			req := &pilosa.ImportRequest{
+				Index:          indexData.indexName,
+				IndexCreatedAt: indexData.createdAt,
+				Field:          fieldData.fieldName,
+				FieldCreatedAt: fieldData.createdAt,
+				Shard:          0, // ignored when using keys
+			}
+			rowKeys := make([]string, 0, len(rowKeysBase)*nShards)
+			colKeys := make([]string, 0, len(rowKeysBase)*nShards)
+			rowIDs = rowIDs[:0]
+			for shard := uint64(0); shard < nShards; shard++ {
+				for i := range rowKeysBase {
+					colKeys = append(colKeys, fmt.Sprintf("s%d-%s", shard, colKeysBase[i]))
+					if keyedField {
+						rowKeys = append(rowKeys, rowKeysBase[i])
+					} else {
+						rowIDs = append(rowIDs, uint64(i))
+					}
+				}
+			}
+			req.ColumnKeys = colKeys
+			if keyedField {
+				req.RowKeys = rowKeys
+			} else {
+				req.RowIDs = rowIDs
+			}
+			var id string
+			var node *test.Command
+			for id, node = range nodesByID {
+				break
+			}
+			if err := node.API.Import(ctx, qcxsByID[id], req); err != nil {
+				t.Fatalf("importing data: %v", err)
+			}
+			expected, err := node.API.FindIndexKeys(ctx, indexData.indexName, colKeys...)
+			if err != nil {
+				t.Fatalf("looking up index keys: %v", err)
+			}
+			for key, id := range expected {
+				// CorruptAMutex should only corrupt things in position 1 of their
+				// shards...
+				if id%(1<<shardwidth.Exponent) != 1 {
+					delete(expected, key)
+				}
+			}
+			if keyedField {
+				fieldValues, err := node.API.FindFieldKeys(ctx, indexData.indexName, fieldData.fieldName, rowKeys...)
+				if err != nil {
+					t.Fatalf("looking up field keys: %v", err)
+				}
+				// Figure out which key got the value 3, delete any records
+				// which would have had that key, because they won't be
+				// conflicts.
+				for key, value := range fieldValues {
+					if value == 3 {
+						for offset, baseKey := range rowKeysBase {
+							if baseKey == key {
+								for i := offset; i < len(rowKeys); i += len(rowKeysBase) {
+									delete(expected, colKeys[i])
+								}
+							}
+						}
+					}
+				}
+			} else {
+				// we set rowKeys to 0-1-2-... for rowKeysBase items, which
+				// tells us which keys we expect to be 3 already.
+				for i := 3; i < len(rowIDs); i += len(rowKeysBase) {
+					delete(expected, colKeys[i])
+				}
+			}
+			// and then we break the mutex and close the Qcxs
+			for id, node := range nodesByID {
+				field, err := node.API.Field(ctx, indexData.indexName, fieldData.fieldName)
+				if err != nil {
+					t.Fatalf("requesting field %s from node %s: %v", fieldData.fieldName, id, err)
+				}
+				pilosa.CorruptAMutex(t, field, qcxsByID[id])
+				err = qcxsByID[id].Finish()
+				if err != nil {
+					t.Fatalf("closing out transaction on node %s: %v", id, err)
+				}
+			}
+			qcx := m0.API.Txf().NewQcx()
+			defer qcx.Abort()
+
+			results, err := m0.API.MutexCheck(ctx, qcx, indexData.indexName, fieldData.fieldName)
+			if err != nil {
+				t.Fatalf("checking mutexes: %v", err)
+			}
+			if keyedField {
+				// this just sorta comes out this way with our hashing; these are
+				// the things which were in position 1 of their shards, and did
+				// not have a value which happens to map to 3.
+				mapped, ok := results.(map[string][]string)
+				if !ok {
+					t.Fatalf("expected map[string][]string, got %T", results)
+				}
+				seen := 0
+				for k, v := range mapped {
+					seen++
+					if _, ok := expected[k]; !ok {
+						t.Fatalf("unexpected collision on key %q", k)
+					}
+					if len(v) != 2 {
+						t.Fatalf("expected exactly two collisions")
+					}
+				}
+				if seen != len(expected) {
+					t.Fatalf("expected exactly %d records to have collisions, got %d", len(expected), seen)
+				}
+			} else {
+				mapped, ok := results.(map[string][]uint64)
+				if !ok {
+					t.Fatalf("expected map[string][]uint64, got %T", results)
+				}
+				seen := 0
+				for k, v := range mapped {
+					seen++
+					if _, ok := expected[k]; !ok {
+						t.Fatalf("unexpected collision on key %q", k)
+					}
+					if len(v) != 2 {
+						t.Fatalf("expected exactly two collisions")
+					}
+				}
+				if seen != len(expected) {
+					t.Fatalf("expected exactly %d records to have collisions, got %d", len(expected), seen)
+				}
+			}
+		})
+	}
+}

client.go

@@ -80,6 +80,7 @@ type InternalClient interface {
 	RetrieveTranslatePartitionFromURI(ctx context.Context, index string, partition int, uri pnet.URI) (io.ReadCloser, error)
 	ImportRoaring(ctx context.Context, uri *pnet.URI, index, field string, shard uint64, remote bool, req *ImportRoaringRequest) error
 	ShardReader(ctx context.Context, index string, shard uint64) (io.ReadCloser, error)
+	MutexCheck(ctx context.Context, uri *pnet.URI, index string, field string) (map[uint64]map[uint64][]uint64, error)
 
 	IDAllocDataReader(ctx context.Context) (io.ReadCloser, error)
 	IndexTranslateDataReader(ctx context.Context, index string, partitionID int) (io.ReadCloser, error)
@@ -213,6 +214,10 @@ func (n nopInternalClient) ImportRoaring(ctx context.Context, uri *pnet.URI, ind
 	return nil
 }
 
+func (n nopInternalClient) MutexCheck(ctx context.Context, uri *pnet.URI, index, field string) (map[uint64]map[uint64][]uint64, error) {
+	return nil, nil
+}
+
 func (n nopInternalClient) ShardReader(ctx context.Context, index string, shard uint64) (io.ReadCloser, error) {
 	return nil, nil
 }

field.go

@@ -1092,6 +1092,23 @@ func (f *Field) Row(tx Tx, rowID uint64) (*Row, error) {
 	}
 }
 
+// mutexCheck performs a sanity-check on the available fragments for a
+// field. The return is map[column]map[shard][]values for collisions only.
+func (f *Field) MutexCheck(ctx context.Context, qcx *Qcx) (map[uint64]map[uint64][]uint64, error) {
+	if f.Type() != FieldTypeMutex {
+		return nil, errors.New("mutex check only valid for mutex fields")
+	}
+	f.mu.RLock()
+	defer f.mu.RUnlock()
+	standard := f.viewMap[viewStandard]
+	if standard == nil {
+		// no standard view present means we've never needed to create it,
+		// so it has no bits set, so it has no extra bits set.
+		return nil, nil
+	}
+	return standard.mutexCheck(ctx, qcx)
+}
+
 // SetBit sets a bit on a view within the field.
 func (f *Field) SetBit(tx Tx, rowID, colID uint64, t *time.Time) (changed bool, err error) {
 	viewName := viewStandard
@@ -1525,7 +1542,7 @@ func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []int64,
 		// we have. But we don't want to allocate four strings per entry, or
 		// recompute and recreate the entire string. We know that only the
 		// YYYYMMDDHH part of the string changes over time.
-		timeStringBuf = make([]byte, len(viewStandard) + 11)
+		timeStringBuf = make([]byte, len(viewStandard)+11)
 		copy(timeStringBuf, []byte(viewStandard))
 		copy(timeStringBuf[len(viewStandard):], []byte("_YYYYMMDDHH"))
 		// Now we have a buffer that contains

field_internal_test.go

@@ -27,10 +27,43 @@ import (
 
 	"github.com/molecula/featurebase/v2/pql"
 	"github.com/molecula/featurebase/v2/roaring"
+	"github.com/molecula/featurebase/v2/shardwidth"
 	"github.com/molecula/featurebase/v2/testhook"
 	. "github.com/molecula/featurebase/v2/vprint" // nolint:staticcheck
 )
 
+// CorruptAMutex breaks a mutex in order to test the mutex-corruption stuff.
+// Note the horrible crime here: This is an exported function which exists only
+// in test builds. This is so that external tests, which aren't inside this
+// package, can call this exported function, and thus get access to functionality
+// that would otherwise not be available to them.
+//
+// This always sets row 3 in column 0 of each shard it finds. Populate the
+// field with existing shards first.
+func CorruptAMutex(tb testing.TB, field *Field, qcx *Qcx) {
+	v := field.view(viewStandard)
+	if v == nil {
+		tb.Fatalf("creating view failed")
+	}
+	frags := v.allFragments()
+	for _, frag := range frags {
+		func() {
+			tx, finisher, err := qcx.GetTx(Txo{Write: true, Index: field.idx, Shard: frag.shard})
+			defer finisher(&err)
+			if err != nil {
+				tb.Fatalf("getting tx: %v", err)
+			}
+			// set a bonus bit, bypassing the mutex handling
+			frag.mu.Lock()
+			_, err = frag.unprotectedSetBit(tx, 3, (frag.shard<<shardwidth.Exponent)+1)
+			frag.mu.Unlock()
+			if err != nil {
+				tb.Fatalf("setting bit: %v", err)
+			}
+		}()
+	}
+}
+
 // Ensure a bsiGroup can adjust to its baseValue.
 func TestBSIGroup_BaseValue(t *testing.T) {
 	b0 := &bsiGroup{

fragment.go

@@ -599,6 +599,17 @@ func (f *fragment) closeStorage() error {
 	return nil
 }
 
+// mutexCheck checks for any entries in fragment which violate the mutex
+// property of having only one value set for a given column ID.
+func (f *fragment) mutexCheck(tx Tx) (map[uint64][]uint64, error) {
+	dup := roaring.NewBitmapMutexDupFilter(f.shard << shardwidth.Exponent)
+	err := tx.ApplyFilter(f.index(), f.field(), f.view(), f.shard, 0, dup)
+	if err != nil {
+		return nil, err
+	}
+	return dup.Report(), nil
+}
+
 // row returns a row by ID.
 func (f *fragment) row(tx Tx, rowID uint64) (*Row, error) {
 	f.mu.Lock()

http/client.go

@@ -284,6 +284,35 @@ func (c *InternalClient) IngestOperations(ctx context.Context, uri *pnet.URI, in
 	return nil
 }
 
+// MutexCheck uses the mutex-check endpoint to request mutex collision data
+// from a single node. It produces per-shard results, and does not translate
+// them.
+func (c *InternalClient) MutexCheck(ctx context.Context, uri *pnet.URI, indexName string, fieldName string) (map[uint64]map[uint64][]uint64, error) {
+	if uri == nil {
+		uri = c.defaultURI
+	}
+	u := uri.Path(fmt.Sprintf("/internal/index/%s/field/%s/mutex-check", indexName, fieldName))
+	req, err := http.NewRequest("GET", u, nil)
+	if err != nil {
+		return nil, errors.Wrap(err, "creating request")
+	}
+	req.Header.Set("Accept", "application/json")
+	req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
+
+	resp, err := c.executeRequest(req.WithContext(ctx))
+	if err != nil {
+		return nil, errors.Wrap(err, "executing request")
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return nil, errors.Errorf("unexpected status code: %s", resp.Status)
+	}
+	var out map[uint64]map[uint64][]uint64
+	dec := json.NewDecoder(resp.Body)
+	err = dec.Decode(&out)
+	return out, err
+}
+
 func (c *InternalClient) PostSchema(ctx context.Context, uri *pnet.URI, s *pilosa.Schema, remote bool) error {
 	u := uri.Path(fmt.Sprintf("/schema?remote=%v", remote))
 	buf, err := json.Marshal(s)

http/handler.go

@@ -389,6 +389,7 @@ func newRouter(handler *Handler) http.Handler {
 	router.HandleFunc("/index/{index}/field/{field}", handler.handlePostField).Methods("POST").Name("PostField")
 	router.HandleFunc("/index/{index}/field/{field}", handler.handleDeleteField).Methods("DELETE").Name("DeleteField")
 	router.HandleFunc("/index/{index}/field/{field}/import", handler.handlePostImport).Methods("POST").Name("PostImport")
+	router.HandleFunc("/index/{index}/field/{field}/mutex-check", handler.handleGetMutexCheck).Methods("GET").Name("GetMutexCheck")
 	router.HandleFunc("/index/{index}/field/{field}/import-roaring/{shard}", handler.handlePostImportRoaring).Methods("POST").Name("PostImportRoaring")
 	router.HandleFunc("/index/{index}/query", handler.handlePostQuery).Methods("POST").Name("PostQuery")
 	router.HandleFunc("/info", handler.handleGetInfo).Methods("GET").Name("GetInfo")
@@ -425,6 +426,7 @@ func newRouter(handler *Handler) http.Handler {
 	router.HandleFunc("/internal/translate/data", handler.handlePostTranslateData).Methods("POST").Name("PostTranslateData")
 	router.HandleFunc("/internal/translate/keys", handler.handlePostTranslateKeys).Methods("POST").Name("PostTranslateKeys")
 	router.HandleFunc("/internal/translate/ids", handler.handlePostTranslateIDs).Methods("POST").Name("PostTranslateIDs")
+	router.HandleFunc("/internal/index/{index}/field/{field}/mutex-check", handler.handleInternalGetMutexCheck).Methods("GET").Name("InternalGetMutexCheck")
 	router.HandleFunc("/internal/index/{index}/field/{field}/remote-available-shards/{shardID}", handler.handleDeleteRemoteAvailableShard).Methods("DELETE")
 	router.HandleFunc("/internal/index/{index}/shard/{shard}/snapshot", handler.handleGetIndexShardSnapshot).Methods("GET").Name("GetIndexShardSnapshot")
 	router.HandleFunc("/internal/index/{index}/shards", handler.handleGetIndexAvailableShards).Methods("GET").Name("GetIndexAvailableShards")
@@ -1621,7 +1623,7 @@ func (h *Handler) handleIngestSchema(w http.ResponseWriter, r *http.Request) {
 	// using otherwise (ironically, to indicate an error)
 	var mapBody []byte
 	var err error
-  if mapBody, err = json.Marshal(cleanupIndexes); err != nil {
+	if mapBody, err = json.Marshal(cleanupIndexes); err != nil {
 		resp.write(w, err)
 	}
 	if _, err = w.Write(mapBody); err != nil {
@@ -2754,6 +2756,58 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+// handleGetMutexCheck handles /mutex-check requests.
+func (h *Handler) handleGetMutexCheck(w http.ResponseWriter, r *http.Request) {
+	if !validHeaderAcceptJSON(r.Header) {
+		http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
+		return
+	}
+	// Get index and field type to determine how to handle the
+	// import data.
+	indexName, fieldName := mux.Vars(r)["index"], mux.Vars(r)["field"]
+	qcx := h.api.Txf().NewQcx()
+	defer qcx.Abort()
+	out, err := h.api.MutexCheck(r.Context(), qcx, indexName, fieldName)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	outBytes, err := json.Marshal(out)
+	if err != nil {
+		http.Error(w, fmt.Sprintf("marshalling response: %v", err), http.StatusInternalServerError)
+	}
+	_, err = w.Write(outBytes)
+	if err != nil {
+		h.logger.Errorf("writing mutex-check response: %v", err)
+	}
+}
+
+// handleInternalGetMutexCheck handles internal (non-forwarding )/mutex-check requests.
+func (h *Handler) handleInternalGetMutexCheck(w http.ResponseWriter, r *http.Request) {
+	if !validHeaderAcceptJSON(r.Header) {
+		http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
+		return
+	}
+	// Get index and field type to determine how to handle the
+	// import data.
+	indexName, fieldName := mux.Vars(r)["index"], mux.Vars(r)["field"]
+	qcx := h.api.Txf().NewQcx()
+	defer qcx.Abort()
+	out, err := h.api.MutexCheckNode(r.Context(), qcx, indexName, fieldName)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	outBytes, err := json.Marshal(out)
+	if err != nil {
+		http.Error(w, fmt.Sprintf("marshalling response: %v", err), http.StatusInternalServerError)
+	}
+	_, err = w.Write(outBytes)
+	if err != nil {
+		h.logger.Errorf("writing mutex-check response: %v", err)
+	}
+}
+
 // handlePostImportRoaring
 func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request) {
 	// Verify that request is only communicating over protobufs.

roaring/filter.go

@@ -755,6 +755,75 @@ func NewBitmapRangeFilter(min, max FilterKey, keyCallback func(FilterKey, int32)
 	return &BitmapRangeFilter{min: min, max: max, kcb: keyCallback, dcb: dataCallback}
 }
 
+// BitmapMutexDupFilter is a filter which identifies cases where the same
+// position has a bit set in more than one row.
+//
+// We keep a slice of the first value seen for every row, with ^0 as the
+// default; when that's already set, things get appended to the entries in
+// the map. At the end, for each entry in the map, we also add its first
+// value to it. Thus, the map holds all the entries, but we're only using
+// the map in the (hopefully rarer) cases where there's duplicate values.
+//
+// The slice is local-coordinates (first column 0), but the map is global
+// coordinates (first column is whatever base was).
+type BitmapMutexDupFilter struct {
+	base  uint64              // the offset of 0 for this, used to accommodate shard offsets
+	extra map[uint64][]uint64 // extra values observed
+	first []uint64            // first values observed
+}
+
+var _ BitmapFilter = &BitmapMutexDupFilter{}
+
+func NewBitmapMutexDupFilter(base uint64) *BitmapMutexDupFilter {
+	filter := &BitmapMutexDupFilter{
+		base:  base,
+		extra: map[uint64][]uint64{},
+		first: make([]uint64, 1<<shardwidth.Exponent),
+	}
+	for i := range filter.first {
+		filter.first[i] = ^uint64(0)
+	}
+	return filter
+}
+
+func (b *BitmapMutexDupFilter) ConsiderKey(key FilterKey, n int32) FilterResult {
+	if n > 0 {
+		return key.NeedData()
+	}
+	return key.RejectOne()
+}
+
+func (b *BitmapMutexDupFilter) ConsiderData(key FilterKey, data *Container) FilterResult {
+	value, basePos := uint64(key)>>rowExponent, uint64(key&keyMask)<<16
+	containerCallback(data, func(u uint16) {
+		pos := basePos + uint64(u)
+		if b.first[pos] != ^uint64(0) {
+			b.extra[pos+b.base] = append(b.extra[pos+b.base], value)
+		} else {
+			b.first[pos] = value
+		}
+	})
+	return key.MatchOne()
+}
+
+// Report returns the set of duplicate values identified.
+func (b *BitmapMutexDupFilter) Report() map[uint64][]uint64 {
+	// copy values into extra, and remove them from first, so calling
+	// Report() again won't cause double-appends.
+	for k, v := range b.extra {
+		kpos := k % (1 << shardwidth.Exponent)
+		if b.first[kpos] != ^uint64(0) {
+			v = append(v, 0)
+			// prepend so the lowest value goes at the beginning
+			copy(v[1:], v[:])
+			v[0] = b.first[kpos]
+			b.first[kpos] = ^uint64(0)
+			b.extra[k] = v
+		}
+	}
+	return b.extra
+}
+
 // ApplyFilterToIterator is a simplistic implementation that applies a bitmap
 // filter to a ContainerIterator, returning an error if it encounters an error.
 //

roaring/filter_internal_test.go

@@ -347,5 +347,51 @@ func TestFilterWithRows(t *testing.T) {
 			}
 		})
 	}
-
+}
+
+func TestMutexDupFilter(t *testing.T) {
+	tests := []struct{
+		pairs [][2]uint64
+		expect map[uint64][]uint64
+	}{
+		{
+			pairs: [][2]uint64{{0, 0}, {1, 0}, {0, 1}},
+			expect: map[uint64][]uint64{0: {0, 1}},
+		},
+		{
+			pairs: [][2]uint64{{0, 0}, {1, 0}, {0, 1}, {0, 2}},
+			expect: map[uint64][]uint64{0: {0, 1, 2}},
+		},
+	}
+	for num, test := range tests {
+		t.Run(fmt.Sprintf("case%d", num), func(t *testing.T) {
+			b := NewSliceBitmap()
+			for _, p := range test.pairs {
+				v := (p[1] << shardwidth.Exponent) | p[0]
+				b.DirectAdd(v)
+			}
+			dup := NewBitmapMutexDupFilter(0)
+			iter, _ := b.Containers.Iterator(0)
+			err := ApplyFilterToIterator(dup, iter)
+			if err != nil {
+				t.Fatalf("applying filter: %v", err)
+			}
+			expected := test.expect
+			got := dup.Report()
+			if len(expected) != len(got) {
+				t.Fatalf("expected %d entries in duplicate map, got %d", len(expected), len(got))
+			}
+			for k, v := range expected {
+				gv := got[k]
+				if len(v) != len(gv) {
+					t.Fatalf("for id %d, expected %d (len %d), got %d (len %d)", k, v, len(v), gv, len(gv))
+				}
+				for j := range v {
+					if gv[j] != v[j] {
+						t.Fatalf("for id %d, expected %d, got %d", k, v[j], gv[j])
+					}
+				}
+			}
+		})
+	}
 }

view.go

@@ -442,6 +442,50 @@ func (v *view) row(txOrig Tx, rowID uint64) (*Row, error) {
 
 }
 
+// mutexCheck checks all available fragments for duplicate values. The return
+// is map[column]map[shard][]values for collisions only.
+func (v *view) mutexCheck(ctx context.Context, qcx *Qcx) (map[uint64]map[uint64][]uint64, error) {
+	// We don't need the context, we just want the context-awareness on the error groups.
+	// It would be nice if the inner functions could use this too...
+	eg, _ := errgroup.WithContext(ctx)
+	throttle := make(chan struct{}, runtime.NumCPU())
+	frags := v.allFragments()
+	results := make([]map[uint64][]uint64, len(frags))
+	for i, frag := range frags {
+		// local copies for the goroutine to use
+		i, frag := i, frag
+		eg.Go(func() error {
+			// limit simultaneous parallel goroutines associated with this
+			throttle <- struct{}{}
+			defer func() {
+				<-throttle
+			}()
+			tx, finisher, err := qcx.GetTx(Txo{Index: v.idx, Shard: frag.shard})
+			if err != nil {
+				return err
+			}
+			defer finisher(&err)
+			results[i], err = frag.mutexCheck(tx)
+			if err != nil {
+				return err
+			}
+			return nil
+		})
+	}
+	err := eg.Wait()
+	if err != nil {
+		return nil, err
+	}
+	out := map[uint64]map[uint64][]uint64{}
+	for i, result := range results {
+		if len(result) == 0 {
+			continue
+		}
+		out[frags[i].shard] = result
+	}
+	return out, nil
+}
+
 // setBit sets a bit within the view.
 func (v *view) setBit(txOrig Tx, rowID, columnID uint64) (changed bool, err error) {
 	shard := columnID / ShardWidth