Add /debug/rbf endpoint for debugging

api.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/molecula/featurebase/v2/disco"
 	"github.com/molecula/featurebase/v2/ingest"
+	"github.com/molecula/featurebase/v2/rbf"
 
 	//"github.com/molecula/featurebase/v2/pg"
 	"github.com/molecula/featurebase/v2/pql"
@@ -3156,6 +3157,21 @@ func (api *API) Plan(ctx context.Context, q string) (*Stmt, error) {
 	return api.server.PlanSQL(ctx, q)
 }
 
+func (api *API) RBFDebugInfo() map[string]*rbf.DebugInfo {
+	infos := make(map[string]*rbf.DebugInfo)
+
+	for key, dbShard := range api.holder.Txf().dbPerShard.Flatmap {
+		wrapper, ok := dbShard.W.(*RbfDBWrapper)
+		if !ok {
+			continue
+		}
+
+		skey := fmt.Sprintf("%s/%d", key.index, key.shard)
+		infos[skey] = wrapper.db.DebugInfo()
+	}
+	return infos
+}
+
 type serverInfo struct {
 	ShardWidth       uint64 `json:"shardWidth"`
 	ReplicaN         int    `json:"replicaN"`

api_test.go

@@ -1415,3 +1415,25 @@ func TestVariousApiTranslateCalls(t *testing.T) {
 		*/
 	}
 }
+
+func TestAPI_RBFDebugInfo(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	c := test.MustRunCluster(t, 1,
+		[]server.CommandOption{
+			server.OptCommandServerOptions(
+				pilosa.OptServerNodeID("node0"),
+				pilosa.OptServerClusterHasher(&offsetModHasher{}),
+				pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
+			)},
+	)
+	defer c.Close()
+
+	coord := c.GetPrimary()
+
+	if _, err := coord.API.CreateIndex(ctx, "i", pilosa.IndexOptions{}); err != nil {
+		t.Fatal(err)
+	} else if infos := coord.API.RBFDebugInfo(); infos == nil {
+		t.Fatal("expected info")
+	}
+}

http/handler.go

@@ -441,6 +441,9 @@ func newRouter(handler *Handler) http.Handler {
 	router.HandleFunc("/internal/idalloc/data", handler.handleIDAllocData).Methods("GET").Name("IDAllocData")
 
 	router.HandleFunc("/internal/restore/{index}/{shardID}", handler.handlePostRestore).Methods("POST").Name("Restore")
+
+	router.HandleFunc("/internal/debug/rbf", handler.handleGetInternalDebugRBFJSON).Methods("GET").Name("GetInternalDebugRBFJSON")
+
 	// endpoints for collecting cpu profiles from a chosen begin point to
 	// when the client wants to stop. Used for profiling imports that
 	// could be long or short.
@@ -2064,6 +2067,18 @@ func validateProtobufHeader(r *http.Request) (error string, code int) {
 	return
 }
 
+// handleGetInternalDebugRBFJSON handles /internal/debug/rbf requests.
+func (h *Handler) handleGetInternalDebugRBFJSON(w http.ResponseWriter, r *http.Request) {
+	buf, err := json.MarshalIndent(h.api.RBFDebugInfo(), "", "  ")
+	if err != nil {
+		http.Error(w, "marshal json: "+err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	w.Write(buf)
+}
+
 // handleGetMetricsJSON handles /metrics.json requests, translating text metrics results to more consumable JSON.
 func (h *Handler) handleGetMetricsJSON(w http.ResponseWriter, r *http.Request) {
 	if !validHeaderAcceptJSON(r.Header) {

rbf/db.go

@@ -7,6 +7,8 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"runtime/debug"
+	"sort"
 	"sync"
 	"syscall"
 
@@ -637,6 +639,7 @@ func (db *DB) Begin(writable bool) (_ *Tx, err error) {
 		pageMap:     db.pageMap,
 		walPageN:    db.walPageN,
 		writable:    writable,
+		stack:       debug.Stack(), // DEBUG
 
 		DeleteEmptyContainer: true,
 	}
@@ -815,6 +818,20 @@ func (db *DB) getCursor(tx *Tx) *Cursor {
 	return c
 }
 
+func (db *DB) DebugInfo() *DebugInfo {
+	info := &DebugInfo{Path: db.Path}
+	for tx := range db.txs {
+		info.Txs = append(info.Txs, tx.DebugInfo())
+	}
+	sort.Slice(info.Txs, func(i, j int) bool { return info.Txs[i].Ptr < info.Txs[j].Ptr })
+	return info
+}
+
+type DebugInfo struct {
+	Path string         `json:"path"`
+	Txs  []*TxDebugInfo `json:"txs"`
+}
+
 // Shared pool for in-memory database pages.
 // These are used before being flushed to disk.
 var pagePool = &sync.Pool{

rbf/db_test.go

@@ -339,6 +339,21 @@ func TestDB_MultiTx(t *testing.T) {
 	}
 }
 
+func TestDB_DebugInfo(t *testing.T) {
+	db := MustOpenDB(t)
+	defer MustCloseDB(t, db)
+
+	tx := MustBegin(t, db, true)
+	defer tx.Rollback()
+
+	info := db.DebugInfo()
+	if got, want := info.Path, db.Path; got != want {
+		t.Fatalf("Path=%q, want %q", got, want)
+	} else if got, want := len(info.Txs), 1; got != want {
+		t.Fatalf("len(Txs)=%d, want %d", got, want)
+	}
+}
+
 // premake pool of random values
 const randPool = (1 << 18)
 

rbf/tx.go

@@ -65,6 +65,9 @@ type Tx struct {
 	// manages to trigger a *deallocation* (which I don't think should be
 	// happening), we'll process that one after the current list is processed.
 	pendingFreelistAdds []uint32
+
+	// DEBUG
+	stack []byte
 }
 
 func (tx *Tx) DBPath() string {
@@ -2042,6 +2045,20 @@ func (tx *Tx) GetSortedFieldViewList() (fvs []txkey.FieldView, _ error) {
 	return
 }
 
+func (tx *Tx) DebugInfo() *TxDebugInfo {
+	return &TxDebugInfo{
+		Ptr:      fmt.Sprintf("%p", tx),
+		Writable: tx.writable,
+		Stack:    string(tx.stack),
+	}
+}
+
+type TxDebugInfo struct {
+	Ptr      string `json:"ptr"`
+	Writable bool   `json:"writable"`
+	Stack    string `json:"stack,omitempty"`
+}
+
 // SnapshotReader returns a reader that provides a snapshot for the current database state.
 func (tx *Tx) SnapshotReader() (io.Reader, error) {
 	if tx.db == nil {