backup mutex sanity check (1681)

api.go

@@ -2188,6 +2188,138 @@ func (api *API) TranslateFieldDB(ctx context.Context, indexName, fieldName strin
 	_, err := store.ReadFrom(rd)
 	return err
 }
+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)
+}
+
+// mergeMutexCollisions adds collisions to an existing map if they aren't already
+// present. It modifies dst.
+func mergeMutexCollisions(dst, src map[uint64][]uint64) {
+	for k, v := range src {
+		if len(v) == 0 {
+			continue
+		}
+		existing := dst[k]
+		if len(existing) == 0 {
+			dst[k] = v
+			continue
+		}
+		// existing and v are both non-empty lists of collisions for this key.
+		// but if replication is working, both nodes should have the SAME list
+		// of collisions, so it's probably worth special-casing that check:
+		if len(existing) == len(v) {
+			different := false
+			for i := range existing {
+				if v[i] != existing[i] {
+					different = true
+					break
+				}
+			}
+			// yay, we can just ignore this
+			if !different {
+				continue
+			}
+		}
+		// combine...
+		existing = append(existing, v...)
+		// sort...
+		sort.Slice(existing, func(i, j int) bool {
+			return existing[i] < existing[j]
+		})
+		// dedup.
+		n := 0
+		prev := existing[0]
+		for i := 0; i < len(existing); i++ {
+			if existing[i] != prev {
+				existing[n] = existing[i]
+				n++
+			}
+			prev = existing[i]
+		}
+		dst[k] = existing[:n]
+	}
+}
+
+// MutexCheck checks for collisions in a given mutex field. The response is
+// a map[shard]map[column]values.
+func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fieldName string, remote bool) (map[uint64]map[uint64][]uint64, error) {
+	if err := api.validate(apiMutexCheck); err != nil {
+		return nil, errors.Wrap(err, "validating api method")
+	}
+	// short path: if this is an internal remote request, only try to solve the
+	// question for these shards.
+	if remote {
+		out, err := api.mutexCheckThisNode(ctx, qcx, indexName, fieldName)
+		return out, err
+	}
+	var nodes []*Node
+	if !remote {
+		nodes = Nodes(api.cluster.nodes).Clone()
+	} else {
+		nodes = []*Node{api.cluster.nodeByID(api.server.nodeID)}
+	}
+
+	/*
+		// 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.server.nodeID
+	results := make([]map[uint64]map[uint64][]uint64, len(nodes))
+	for i, node := range 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
+	}
+	var out map[uint64]map[uint64][]uint64
+	for _, nodeResults := range results {
+		if len(nodeResults) == 0 {
+			continue
+		}
+		if out == nil {
+			out = nodeResults
+			continue
+		}
+		for k, v := range nodeResults {
+			if len(v) == 0 {
+				continue
+			}
+			existing := out[k]
+			// results from two nodes. might be normal with replication.
+			if len(existing) == 0 {
+				out[k] = v
+				continue
+			}
+			// we have two maps for this shard. whee.
+			mergeMutexCollisions(existing, v)
+			// we don't store it back into out[k] because it was already
+			// a non-empty map, so stores to it update it. yay?
+		}
+	}
+	return out, nil
+}
 
 type serverInfo struct {
 	ShardWidth       uint64 `json:"shardWidth"`
@@ -2249,11 +2381,13 @@ const (
 	apiIDReserve
 	apiIDCommit
 	apiIDReset
+	apiMutexCheck //Backported from 1681 Caution
 )
 
 var methodsCommon = map[apiMethod]struct{}{
 	apiClusterMessage: {},
 	apiSetCoordinator: {},
+	apiMutexCheck:     {},
 }
 
 var methodsResizing = map[apiMethod]struct{}{

client.go

@@ -84,6 +84,8 @@ type InternalClient interface {
 
 	GetNodeUsage(ctx context.Context, uri *URI) (map[string]NodeUsage, error)
 	GetPastQueries(ctx context.Context, uri *URI) ([]PastQueryStatus, error)
+
+	MutexCheck(ctx context.Context, uri *URI, index string, field string) (map[uint64]map[uint64][]uint64, error)
 }
 
 //===============
@@ -251,3 +253,7 @@ func (n nopInternalClient) GetNodeUsage(ctx context.Context, uri *URI) (map[stri
 func (n nopInternalClient) GetPastQueries(ctx context.Context, uri *URI) ([]PastQueryStatus, error) {
 	return nil, nil
 }
+
+func (n nopInternalClient) MutexCheck(ctx context.Context, uri *URI, index, field string) (map[uint64]map[uint64][]uint64, error) {
+	return nil, nil
+}

field.go

@@ -1229,6 +1229,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

field_internal_test.go

@@ -27,6 +27,7 @@ import (
 
 	"github.com/pilosa/pilosa/v2/pql"
 	"github.com/pilosa/pilosa/v2/roaring"
+	"github.com/pilosa/pilosa/v2/shardwidth"
 	"github.com/pilosa/pilosa/v2/testhook"
 )
 
@@ -922,3 +923,27 @@ func TestBSIGroup_TxReopenDB(t *testing.T) {
 	// the test: can we re-open a BSI fragment under Tx store
 	_ = f.Reopen()
 }
+
+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)
+			frag.mu.Unlock()
+			if err != nil {
+				tb.Fatalf("setting bit: %v", err)
+			}
+		}()
+	}
+}

fragment.go

@@ -585,6 +585,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

@@ -133,6 +133,34 @@ func (c *InternalClient) Schema(ctx context.Context) ([]*pilosa.IndexInfo, error
 	return rsp.Indexes, nil
 }
 
+// MutexCheck uses the mutex-check endpoint to request mutex collision data
+// from a single node.
+func (c *InternalClient) MutexCheck(ctx context.Context, uri *pilosa.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 *pilosa.URI, s *pilosa.Schema, remote bool) error {
 	u := uri.Path(fmt.Sprintf("/schema?remote=%v", remote))
 	buf, err := json.Marshal(s)

http/handler.go

@@ -386,6 +386,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")
@@ -422,6 +423,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}/attr/diff", handler.handlePostFieldAttrDiff).Methods("POST").Name("PostFieldAttrDiff")
 	router.HandleFunc("/internal/index/{index}/field/{field}/remote-available-shards/{shardID}", handler.handleDeleteRemoteAvailableShard).Methods("DELETE")
 	router.HandleFunc("/internal/nodes", handler.handleGetNodes).Methods("GET").Name("GetNodes")
@@ -2479,6 +2481,58 @@ func (h *Handler) handlePostImportColumnAttrs(w http.ResponseWriter, r *http.Req
 	}
 }
 
+// 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, false)
+	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.Printf("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.MutexCheck(r.Context(), qcx, indexName, fieldName, true)
+	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.Printf("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

@@ -349,3 +349,50 @@ 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

@@ -447,6 +447,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