Skip to content

Commit 7761a3a

Browse files
rohrerjrohrerj
committed
update to statistics endpoint to fix wrong computation when assets had been combined
1 parent e88ebd2 commit 7761a3a

8 files changed

Lines changed: 162 additions & 53 deletions

File tree

‎marketplace/db/schema.sql‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -76,4 +76,17 @@ CREATE TABLE IF NOT EXISTS Redemption_Delegations(
7676
encodings BLOB NOT NULL,
7777
PRIMARY KEY(isd_id, as_id)
7878
);
79-
CREATE INDEX IF NOT EXISTS idx_redemption_expiration ON Redemption_Delegations(expiration);
79+
CREATE INDEX IF NOT EXISTS idx_redemption_expiration ON Redemption_Delegations(expiration);
80+
CREATE TABLE IF NOT EXISTS Asset_Events(
81+
isd_id INTEGER NOT NULL,
82+
as_id INTEGER NOT NULL,
83+
ingress INTEGER,
84+
egress INTEGER,
85+
bandwidth INTEGER NOT NULL,
86+
starts_at TEXT NOT NULL,
87+
stops_at TEXT NOT NULL,
88+
price INTEGER NOT NULL,
89+
event_type INTEGER NOT NULL,
90+
event_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
91+
);
92+
CREATE INDEX IF NOT EXISTS idx_asset_events ON Asset_Events(event_type, isd_id, as_id, stops_at, starts_at);

‎marketplace/db/sqlite.go‎

Lines changed: 61 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ type Repository interface {
4343
CreateASUser(ctx context.Context, user *DBASUser) (int64, error)
4444
UpdateAccountMoney(ctx context.Context, id int64, amount int64) (int64, error)
4545
UpdateASMoney(ctx context.Context, ia addr.IA, amount int64) (int64, error)
46-
SearchAssetsForStatistics(ctx context.Context, params *StatisticsQuery) ([]*DBStat, error)
46+
SearchAssetsForStatistics(ctx context.Context, params *StatisticsQuery) ([]*DBStat, []*DBStat, error)
4747
FindUsedReservations(ctx context.Context, params *UsedReservationsQuery) ([]*UsedReservation, error)
4848
CreateOrUpdateRedemptionDelegations(ctx context.Context, r *RedemptionDelegation) (int64, error)
4949
FindRedemptionDelegations(ctx context.Context) ([]*RedemptionDelegation, error)
@@ -62,6 +62,7 @@ type Repository interface {
6262
AssignAsset(ctx context.Context, assetID int64, accountIDFrom int64, accountIDTo int64) (int64, error)
6363
AssignReservation(ctx context.Context, id int64, accountIDFrom int64, accountIDTo int64) (int64, error)
6464
RemoveAsset(ctx context.Context, assetID int64) error
65+
RegisterAssetEvent(ctx context.Context, a *DBAsset, eventType AssetEventType) (int64, error)
6566
TransitionAsset(ctx context.Context, assetID int64, accountID *int64, from AssetState, to AssetState) (*DBAsset, error)
6667
}
6768

@@ -548,53 +549,71 @@ func (e *executor) buildUsedReservationsQuery(params *UsedReservationsQuery) (st
548549
return query, args
549550
}
550551

551-
func (e *executor) SearchAssetsForStatistics(
552-
ctx context.Context,
553-
params *StatisticsQuery,
554-
) ([]*DBStat, error) {
552+
func (e *executor) SearchAssetsForStatistics(ctx context.Context, params *StatisticsQuery) ([]*DBStat, []*DBStat, error) {
555553
if e.read == nil {
556-
return nil, serrors.New("No database open")
554+
return nil, nil, serrors.New("No database open")
557555
}
558-
stmt, args := e.buildStatisticsQuery(params)
559-
rows, err := e.read.QueryContext(ctx, stmt, args...)
556+
publishStmt, args := e.buildAssetEventsQuery(params, AssetPublished)
557+
publishRows, err := e.read.QueryContext(ctx, publishStmt, args...)
560558
if err != nil {
561-
return nil, serrors.New("Error looking up assets", "err", err, "q", stmt)
559+
return nil, nil, serrors.New("Error looking up assets", "err", err, "q", publishStmt)
562560
}
563-
defer rows.Close()
564-
var res []*DBStat
565-
for rows.Next() {
561+
defer publishRows.Close()
562+
var publishedAssets []*DBStat
563+
for publishRows.Next() {
566564
a := &DBStat{}
567565
var startsAtString string
568566
var stopsAtString string
569-
err = rows.Scan(&a.OwnerId, &a.Bandwidth, &a.Price, &startsAtString, &stopsAtString)
567+
err = publishRows.Scan(&a.Bandwidth, &a.Price, &startsAtString, &stopsAtString)
570568
if err != nil {
571-
return nil, serrors.Wrap("Error reading DB response", err)
569+
return nil, nil, serrors.Wrap("Error reading DB response", err)
572570
}
573571
a.StartsAt, err = time.Parse(time.RFC3339, startsAtString)
574572
if err != nil {
575-
return nil, err
573+
return nil, nil, err
576574
}
577575
a.StopsAt, err = time.Parse(time.RFC3339, stopsAtString)
578576
if err != nil {
579-
return nil, err
577+
return nil, nil, err
580578
}
581-
res = append(res, a)
579+
publishedAssets = append(publishedAssets, a)
582580
}
583-
return res, nil
581+
boughtStmt, args := e.buildAssetEventsQuery(params, AssetBought)
582+
boughtRows, err := e.read.QueryContext(ctx, boughtStmt, args...)
583+
if err != nil {
584+
return nil, nil, serrors.New("Error looking up assets", "err", err, "q", boughtStmt)
585+
}
586+
defer boughtRows.Close()
587+
var boughtAssets []*DBStat
588+
for boughtRows.Next() {
589+
a := &DBStat{}
590+
var startsAtString string
591+
var stopsAtString string
592+
err = boughtRows.Scan(&a.Bandwidth, &a.Price, &startsAtString, &stopsAtString)
593+
if err != nil {
594+
return nil, nil, serrors.Wrap("Error reading DB response", err)
595+
}
596+
a.StartsAt, err = time.Parse(time.RFC3339, startsAtString)
597+
if err != nil {
598+
return nil, nil, err
599+
}
600+
a.StopsAt, err = time.Parse(time.RFC3339, stopsAtString)
601+
if err != nil {
602+
return nil, nil, err
603+
}
604+
boughtAssets = append(boughtAssets, a)
605+
}
606+
return publishedAssets, boughtAssets, nil
584607
}
585608

586-
func (e *executor) buildStatisticsQuery(params *StatisticsQuery) (string, []any) {
609+
func (e *executor) buildAssetEventsQuery(params *StatisticsQuery, eventType AssetEventType) (string, []any) {
587610
var args []any
588611
where := []string{}
589612
query := []string{
590-
"SELECT account_id, bandwidth, price, starts_at, stops_at FROM Assets",
591-
}
592-
where = append(where, "(isd_id=?) AND (as_id=?) AND (stops_at > ?) AND (starts_at <= ?)")
593-
args = append(args,
594-
int64(params.IA.ISD()),
595-
int64(params.IA.AS()),
596-
params.WindowStart,
597-
params.WindowEnd)
613+
"SELECT bandwidth, price, starts_at, stops_at FROM Asset_Events",
614+
}
615+
where = append(where, "(event_type = ?) AND (isd_id=?) AND (as_id=?) AND (stops_at > ?) AND (starts_at <= ?)")
616+
args = append(args, eventType, int64(params.IA.ISD()), int64(params.IA.AS()), params.WindowStart, params.WindowEnd)
598617
if params.Ingress != nil {
599618
where = append(where, "(ingress=?)")
600619
args = append(args, *params.Ingress)
@@ -607,6 +626,21 @@ func (e *executor) buildStatisticsQuery(params *StatisticsQuery) (string, []any)
607626
return strings.Join(query, "\n"), args
608627
}
609628

629+
func (e *executor) RegisterAssetEvent(ctx context.Context, a *DBAsset, eventType AssetEventType) (int64, error) {
630+
if e.write == nil {
631+
return 0, serrors.New("No database open")
632+
}
633+
var err error
634+
inst := `INSERT INTO Asset_Events (event_type, isd_id, as_id, ingress, egress, bandwidth, starts_at, stops_at, price)
635+
VALUES(?,?,?,?,?,?,?,?)`
636+
res, err := e.write.ExecContext(ctx, inst, eventType, a.IA.ISD(), a.IA.AS(), a.IfIdIngress, a.IfIdEgress, a.Bandwidth,
637+
a.StartAt.UTC().Format(time.RFC3339), a.StopsAt.UTC().Format(time.RFC3339), a.Price)
638+
if err != nil {
639+
return 0, err
640+
}
641+
return res.LastInsertId()
642+
}
643+
610644
func (e *executor) Search(ctx context.Context, params *AssetQuery) ([]*DBAsset, error) {
611645
if e.read == nil {
612646
return nil, serrors.New("No database open")

‎marketplace/db/types.go‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,13 @@ const (
3535
AssetStateCombinePending
3636
)
3737

38+
type AssetEventType int
39+
40+
const (
41+
AssetPublished AssetEventType = iota
42+
AssetBought
43+
)
44+
3845
// AssetID is the unsigned representation of an asset identifier used by the
3946
// marketplace API.
4047
type AssetID uint64
@@ -104,7 +111,6 @@ type DBStat struct {
104111
Bandwidth int64
105112
StartsAt time.Time
106113
StopsAt time.Time
107-
OwnerId sql.NullInt64
108114
}
109115

110116
type DBReservation struct {

‎marketplace/service.go‎

Lines changed: 42 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@ import (
1818
"context"
1919
"database/sql"
2020
"encoding/binary"
21-
"fmt"
2221
"sync"
2322
"time"
2423

@@ -166,7 +165,6 @@ func (s *Service) SplitAsset(ctx context.Context, req *connect.Request[hummingbi
166165
}
167166

168167
func (s *Service) BuyAssets(ctx context.Context, req *connect.Request[hummingbird.BuyAssetsRequest]) (*connect.Response[hummingbird.BuyAssetsResponse], error) {
169-
fmt.Println("BuyAssets")
170168
user, ok := ctx.Value("user").(int64)
171169
if !ok {
172170
return nil, connect.NewError(connect.CodePermissionDenied, serrors.New("user_id not provided"))
@@ -243,7 +241,6 @@ func (s *Service) FetchReservations(ctx context.Context, req *connect.Request[hu
243241
}
244242

245243
func (s *Service) Info(context.Context, *connect.Request[hummingbird.MarketplaceInfoRequest]) (*connect.Response[hummingbird.MarketplaceInfoResponse], error) {
246-
fmt.Println("Info")
247244
return &connect.Response[hummingbird.MarketplaceInfoResponse]{
248245
Msg: &hummingbird.MarketplaceInfoResponse{
249246
ApiMajorVersion: s.info.ApiMajorVersion,
@@ -263,7 +260,6 @@ func (s *Service) Info(context.Context, *connect.Request[hummingbird.Marketplace
263260
}
264261

265262
func (s *Service) UpdateAssets(ctx context.Context, req *connect.Request[hummingbird.UpdateAssetsRequest]) (*connect.Response[hummingbird.UpdateAssetsResponse], error) {
266-
fmt.Println("UpdateAssets")
267263
ia, ok := ctx.Value("user").(addr.IA)
268264
if !ok {
269265
return nil, connect.NewError(connect.CodePermissionDenied, serrors.New("ia not provided"))
@@ -346,7 +342,6 @@ func (s *Service) UpdateAssets(ctx context.Context, req *connect.Request[humming
346342
}
347343

348344
func (s *Service) PublishAsset(ctx context.Context, req *connect.Request[hummingbird.PublishAssetRequest]) (*connect.Response[hummingbird.PublishAssetResponse], error) {
349-
fmt.Println("PublishAsset")
350345
ia, ok := ctx.Value("user").(addr.IA)
351346
if !ok {
352347
return nil, connect.NewError(connect.CodePermissionDenied, serrors.New("ia not provided"))
@@ -394,7 +389,6 @@ func (s *Service) RedeemAsset(
394389
ctx context.Context,
395390
req *connect.Request[hummingbird.RedeemAssetRequest],
396391
) (*connect.Response[hummingbird.RedeemAssetResponse], error) {
397-
fmt.Println("RedeemAsset")
398392
user, ok := ctx.Value("user").(int64)
399393
if !ok {
400394
return nil, connect.NewError(connect.CodePermissionDenied, serrors.New("user_id not provided"))
@@ -535,23 +529,22 @@ func (s *Service) RedeemAsset(
535529
}
536530

537531
func (s *Service) Statistics(ctx context.Context, req *connect.Request[hummingbird.StatisticsRequest]) (*connect.Response[hummingbird.StatisticsResponse], error) {
538-
fmt.Println("Statistics")
539532
ia, ok := ctx.Value("user").(addr.IA)
540533
if !ok {
541534
return nil, connect.NewError(connect.CodePermissionDenied, serrors.New("ia not provided"))
542535
}
543-
step := hbird.RoundUpDuration(time.Duration(req.Msg.Step)*time.Second, time.Duration(s.info.StatisticsTimeGranularity)*time.Second)
536+
var step time.Duration
537+
step = hbird.RoundUpDuration(time.Duration(req.Msg.Step)*time.Second, time.Duration(s.info.StatisticsTimeGranularity)*time.Second)
544538
windowStart := req.Msg.Start.AsTime().UTC().Truncate(time.Duration(s.info.StatisticsTimeGranularity) * time.Second)
545539
windowEnd := hbird.RoundUpTime(req.Msg.End.AsTime().UTC(), time.Duration(s.info.StatisticsTimeGranularity)*time.Second)
546540
num_intervals := int(windowEnd.Sub(windowStart) / step)
547-
fmt.Println(step, windowStart, windowEnd)
548541
if num_intervals > 1024 {
549542
return nil, connect.NewError(connect.CodeResourceExhausted, serrors.New("too many intervals"))
550543
}
551544
income := make([]uint64, num_intervals)
552545
bwBought := make([]uint64, num_intervals)
553-
bwListed := make([]uint64, num_intervals)
554-
assets, err := s.store.Statistics(ctx, &db.StatisticsQuery{
546+
bwPublished := make([]uint64, num_intervals)
547+
publishedAssets, boughtAssets, err := s.store.Statistics(ctx, &db.StatisticsQuery{
555548
IA: ia,
556549
WindowStart: windowStart.Format(time.RFC3339),
557550
WindowEnd: windowEnd.Format(time.RFC3339),
@@ -561,8 +554,7 @@ func (s *Service) Statistics(ctx context.Context, req *connect.Request[hummingbi
561554
if err != nil {
562555
return nil, connect.NewError(connect.CodeInvalidArgument, err)
563556
}
564-
565-
for _, asset := range assets {
557+
for _, asset := range publishedAssets {
566558
start := asset.StartsAt
567559
stop := asset.StopsAt
568560
if asset.StartsAt.Before(windowStart) {
@@ -593,19 +585,50 @@ func (s *Service) Statistics(ctx context.Context, req *connect.Request[hummingbi
593585
}
594586
duration := uint64(overlapEnd.Sub(overlapStart).Seconds())
595587
bwTimesDuration := uint64(asset.Bandwidth) * duration
596-
if !asset.OwnerId.Valid {
597-
bwListed[i] += bwTimesDuration
598-
} else {
599-
bwBought[i] += bwTimesDuration
600-
income[i] += bwTimesDuration * uint64(asset.Price)
588+
bwPublished[i] += bwTimesDuration
589+
}
590+
}
591+
592+
for _, asset := range boughtAssets {
593+
start := asset.StartsAt
594+
stop := asset.StopsAt
595+
if asset.StartsAt.Before(windowStart) {
596+
start = windowStart
597+
}
598+
if asset.StopsAt.After(windowEnd) {
599+
stop = windowEnd
600+
}
601+
first := int(start.Sub(windowStart) / step)
602+
603+
last := int(stop.Sub(windowStart) / step)
604+
if first < 0 || last > num_intervals {
605+
continue
606+
}
607+
if stop.Equal(windowStart.Add(time.Duration(last) * step)) {
608+
last--
609+
}
610+
for i := first; i <= last && i < num_intervals; i++ {
611+
intervalStart := windowStart.Add(time.Duration(i) * step)
612+
intervalEnd := intervalStart.Add(step)
613+
overlapStart := start
614+
overlapEnd := stop
615+
if asset.StartsAt.Before(intervalStart) {
616+
overlapStart = intervalStart
617+
}
618+
if asset.StopsAt.After(intervalEnd) {
619+
overlapEnd = intervalEnd
601620
}
621+
duration := uint64(overlapEnd.Sub(overlapStart).Seconds())
622+
bwTimesDuration := uint64(asset.Bandwidth) * duration
623+
bwBought[i] += bwTimesDuration
624+
income[i] += bwTimesDuration * uint64(asset.Price)
602625
}
603626
}
604627
respEntries := make([]*hummingbird.StatisticsResponseEntry, num_intervals)
605628
for i := 0; i < num_intervals; i++ {
606629
respEntries[i] = &hummingbird.StatisticsResponseEntry{
607630
Revenue: income[i],
608-
BandwidthUtilization: float64(bwBought[i]) / float64(bwBought[i]+bwListed[i]),
631+
BandwidthUtilization: float64(bwBought[i]) / float64(bwPublished[i]),
609632
}
610633
}
611634
return &connect.Response[hummingbird.StatisticsResponse]{
@@ -616,7 +639,6 @@ func (s *Service) Statistics(ctx context.Context, req *connect.Request[hummingbi
616639
}
617640

618641
func (s *Service) SearchAssets(ctx context.Context, req *connect.Request[hummingbird.SearchAssetsRequest]) (*connect.Response[hummingbird.SearchAssetsResponse], error) {
619-
fmt.Println("SearchAssets")
620642
var owner_id *int64
621643
switch user := ctx.Value("user").(type) {
622644
case int64:

‎marketplace/storage/storage.go‎

Lines changed: 19 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -91,7 +91,19 @@ func (s *MarketplaceStorage) PublishAsset(
9191
if err != nil {
9292
return 0, err
9393
}
94-
return s.db.InsertAsset(ctx, a)
94+
var assetId int64
95+
err = s.db.WithTx(ctx, func(tx marketplacedb.Repository) error {
96+
assetId, err = tx.InsertAsset(ctx, a)
97+
if err != nil {
98+
return err
99+
}
100+
_, err := tx.RegisterAssetEvent(ctx, a, marketplacedb.AssetPublished)
101+
if err != nil {
102+
return err
103+
}
104+
return nil
105+
})
106+
return assetId, err
95107
}
96108
func (s *MarketplaceStorage) UpdateListedAsset(
97109
ctx context.Context,
@@ -550,7 +562,7 @@ func (s *MarketplaceStorage) SplitAsset(
550562
func (s *MarketplaceStorage) Statistics(
551563
ctx context.Context,
552564
params *marketplacedb.StatisticsQuery,
553-
) ([]*marketplacedb.DBStat, error) {
565+
) ([]*marketplacedb.DBStat, []*marketplacedb.DBStat, error) {
554566
return s.db.SearchAssetsForStatistics(ctx, params)
555567
}
556568

@@ -630,6 +642,11 @@ func (s *MarketplaceStorage) BuyAssets(
630642
if err != nil {
631643
return err
632644
}
645+
newAsset.ID = id
646+
_, err = tx.RegisterAssetEvent(ctx, newAsset, marketplacedb.AssetBought)
647+
if err != nil {
648+
return err
649+
}
633650
totalAssetPrice, err := hbird.ReservationPrice(
634651
dbAsset.Price,
635652
split.Split.Bandwidth,

‎pkg/hummingbird/hummingbird.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,9 @@ type RequestMap map[addr.IA]RedemptionRequest
4949
// RoundUpDuration rounds duration up to the next multiple. A non-positive
5050
// multiple leaves duration unchanged.
5151
func RoundUpDuration(duration, multiple time.Duration) time.Duration {
52+
if duration == 0 {
53+
return multiple
54+
}
5255
if multiple <= 0 {
5356
return duration
5457
}

‎pkg/hummingbird/marketplace/client.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -635,7 +635,7 @@ func (c *MarketplaceClient) ObtainReservationsForInterfacePairs(ctx context.Cont
635635
foundAssetsMap := make(map[uint64][]*hummingbird.SearchAsset)
636636
for _, boughtAsset := range boughtAssets {
637637
for _, foundAsset := range foundAssets {
638-
if bytes.Compare(boughtAsset.AssetId, foundAsset.AssetId) == 0 {
638+
if bytes.Equal(boughtAsset.AssetId, foundAsset.AssetId) {
639639
currentSlice, found := foundAssetsMap[foundAsset.Ia]
640640
if found {
641641
foundAssetsMap[foundAsset.Ia] = append(currentSlice, foundAsset)

0 commit comments

Comments
 (0)