package hapi import ( "context" "sort" "time" "github.com/BurntSushi/toml" "golang.org/x/xerrors" "github.com/filecoin-project/go-address" "github.com/filecoin-project/lotus/blockstore" "github.com/filecoin-project/lotus/chain/actors/builtin/miner" "github.com/filecoin-project/lotus/chain/store" "github.com/filecoin-project/lotus/chain/types" ) const watchInterval = time.Second * 10 func (a *app) watchActor() { ticker := time.NewTicker(watchInterval) for { err := a.updateActor(context.TODO()) if err != nil { log.Errorw("updating rpc info", "error", err) } select { case <-ticker.C: } } } type minimalActorInfo struct { Addresses []struct { MinerAddresses []string } } func (a *app) updateActor(ctx context.Context) error { a.rpcInfoLk.Lock() api := a.workingApi a.rpcInfoLk.Unlock() stor := store.ActorStore(ctx, blockstore.NewReadCachedBlockstore(blockstore.NewAPIBlockstore(a.workingApi), ChainBlockCache)) if api == nil { log.Warnw("no working api yet") return nil } var actorInfos []actorInfo confNameToAddr := map[address.Address][]string{} // address -> config names err := forEachConfig[minimalActorInfo](a, func(name string, info minimalActorInfo) error { for _, aset := range info.Addresses { for _, addr := range aset.MinerAddresses { a, err := address.NewFromString(addr) if err != nil { return xerrors.Errorf("parsing address: %w", err) } confNameToAddr[a] = append(confNameToAddr[a], name) } } return nil }) if err != nil { return err } wins, err := a.spWins(ctx) if err != nil { return xerrors.Errorf("getting sp wins: %w", err) } for addr, cnames := range confNameToAddr { p, err := api.StateMinerPower(ctx, addr, types.EmptyTSK) if err != nil { return xerrors.Errorf("getting miner power: %w", err) } dls, err := api.StateMinerDeadlines(ctx, addr, types.EmptyTSK) if err != nil { return xerrors.Errorf("getting deadlines: %w", err) } mact, err := api.StateGetActor(ctx, addr, types.EmptyTSK) if err != nil { return xerrors.Errorf("getting actor: %w", err) } mas, err := miner.Load(stor, mact) if err != nil { return err } outDls := []actorDeadline{} for dlidx := range dls { p, err := api.StateMinerPartitions(ctx, addr, uint64(dlidx), types.EmptyTSK) if err != nil { return xerrors.Errorf("getting partition: %w", err) } dl := actorDeadline{ Empty: false, Current: false, // todo Proven: false, PartFaulty: false, Faulty: false, } var live, faulty uint64 for _, part := range p { l, err := part.LiveSectors.Count() if err != nil { return xerrors.Errorf("getting live sectors: %w", err) } live += l f, err := part.FaultySectors.Count() if err != nil { return xerrors.Errorf("getting faulty sectors: %w", err) } faulty += f } dl.Empty = live == 0 dl.Proven = live > 0 && faulty == 0 dl.PartFaulty = faulty > 0 dl.Faulty = faulty > 0 && faulty == live outDls = append(outDls, dl) } pd, err := api.StateMinerProvingDeadline(ctx, addr, types.EmptyTSK) if err != nil { return xerrors.Errorf("getting proving deadline: %w", err) } if len(outDls) != 48 { return xerrors.Errorf("expected 48 deadlines, got %d", len(outDls)) } outDls[pd.Index].Current = true avail, err := mas.AvailableBalance(mact.Balance) if err != nil { return xerrors.Errorf("getting available balance: %w", err) } mi, err := mas.Info() if err != nil { return xerrors.Errorf("getting miner info: %w", err) } wbal, err := api.WalletBalance(ctx, mi.Worker) if err != nil { return xerrors.Errorf("getting worker balance: %w", err) } actorInfos = append(actorInfos, actorInfo{ Address: addr.String(), CLayers: cnames, QualityAdjustedPower: types.DeciStr(p.MinerPower.QualityAdjPower), RawBytePower: types.DeciStr(p.MinerPower.RawBytePower), Deadlines: outDls, ActorBalance: types.FIL(mact.Balance).Short(), ActorAvailable: types.FIL(avail).Short(), WorkerBalance: types.FIL(wbal).Short(), Win1: wins[addr].Win1, // note: zero values are fine here Win7: wins[addr].Win7, Win30: wins[addr].Win30, }) } sort.Slice(actorInfos, func(i, j int) bool { return actorInfos[i].Address < actorInfos[j].Address }) a.actorInfoLk.Lock() a.actorInfos = actorInfos a.actorInfoLk.Unlock() return nil } func (a *app) loadConfigs(ctx context.Context) (map[string]string, error) { rows, err := a.db.Query(ctx, `SELECT title, config FROM harmony_config`) if err != nil { return nil, xerrors.Errorf("getting db configs: %w", err) } configs := make(map[string]string) for rows.Next() { var title, config string if err := rows.Scan(&title, &config); err != nil { return nil, xerrors.Errorf("scanning db configs: %w", err) } configs[title] = config } return configs, nil } type wins struct { SpID int64 `db:"sp_id"` Win1 int64 `db:"win1"` Win7 int64 `db:"win7"` Win30 int64 `db:"win30"` } func (a *app) spWins(ctx context.Context) (map[address.Address]wins, error) { var w []wins // note: this query uses mining_tasks_won_sp_id_base_compute_time_index err := a.db.Select(ctx, &w, `WITH wins AS ( SELECT sp_id, base_compute_time, won FROM mining_tasks WHERE won = true AND base_compute_time > NOW() - INTERVAL '30 days' ) SELECT sp_id, COUNT(*) FILTER (WHERE base_compute_time > NOW() - INTERVAL '1 day') AS "win1", COUNT(*) FILTER (WHERE base_compute_time > NOW() - INTERVAL '7 days') AS "win7", COUNT(*) FILTER (WHERE base_compute_time > NOW() - INTERVAL '30 days') AS "win30" FROM wins GROUP BY sp_id ORDER BY sp_id`) if err != nil { return nil, xerrors.Errorf("query win counts: %w", err) } wm := make(map[address.Address]wins) for _, wi := range w { ma, err := address.NewIDAddress(uint64(wi.SpID)) if err != nil { return nil, xerrors.Errorf("parsing miner address: %w", err) } wm[ma] = wi } return wm, nil } func forEachConfig[T any](a *app, cb func(name string, v T) error) error { confs, err := a.loadConfigs(context.Background()) if err != nil { return err } for name, tomlStr := range confs { var info T if err := toml.Unmarshal([]byte(tomlStr), &info); err != nil { return xerrors.Errorf("unmarshaling %s config: %w", name, err) } if err := cb(name, info); err != nil { return xerrors.Errorf("cb: %w", err) } } return nil }