Skip to content

Commit d62f55f

Browse files
Cre 291/max spend for step (#17955)
* convert to decimals, update metering mode, refactor tests, provide spending limit * add tests for full test coverage * address comments * fix errors due to change from NullDecimal * make linter happy * move default precision setting to init func due to race conditions * remove precision setting for decimal * add info call to test * address feedback * address feedback * more feedback * Cre 291/metering mode refactor (#18273) * remove metering mode from balance store and consolidate into report struct * pr nits --------- Co-authored-by: patrickhuie19 <patrick.huie@smartcontract.com> * lint --------- Co-authored-by: patrickhuie19 <patrick.huie@smartcontract.com>
1 parent 1f0858d commit d62f55f

8 files changed

Lines changed: 800 additions & 505 deletions

File tree

core/services/workflows/engine.go

Lines changed: 38 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import (
99
"time"
1010

1111
"github.com/jonboulle/clockwork"
12+
"github.com/shopspring/decimal"
1213

1314
"github.com/smartcontractkit/chainlink-common/pkg/aggregation"
1415
"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
@@ -773,19 +774,41 @@ func (e *Engine) workerForStepRequest(ctx context.Context, msg stepRequest) {
773774
Ref: msg.stepRef,
774775
}
775776

777+
curStepID := "UNSET"
778+
curStep, verr := e.workflow.Vertex(msg.stepRef)
779+
if verr == nil {
780+
curStepID = curStep.ID
781+
} else {
782+
l.Errorf("failed to resolve step in workflow; error %v", verr)
783+
}
784+
785+
info, err := curStep.capability.Info(ctx)
786+
if err != nil {
787+
l.Errorf("failed to get capability info: %s", err)
788+
}
789+
790+
spendLimits := []capabilities.SpendLimit{}
791+
776792
meteringReport, meteringOK := e.meterReports.Get(msg.state.ExecutionID)
777793
if meteringOK {
778-
// TODO: https://smartcontract-it.atlassian.net/browse/CRE-477 Get capability info by getting the workflow vertex and talking to the capaiblity
779-
// TODO: https://smartcontract-it.atlassian.net/browse/CRE-285 get max spend per step. Compare to availability and limits.
794+
// TODO: https://smartcontract-it.atlassian.net/browse/CRE-284 parse user max spend for step
795+
userMaxSpend := decimal.NewNullDecimal(decimal.Zero)
796+
userMaxSpend.Valid = false
797+
780798
// NOTE: e.maxWorkerLimit is a static number leading to the availability always being undercut.
781-
availableForCall, err := meteringReport.GetAvailableForInvocation(e.maxWorkerLimit)
799+
spendLimit, err := meteringReport.GetMaxSpendForInvocation(userMaxSpend, e.maxWorkerLimit)
800+
782801
if err != nil {
783802
l.Error(fmt.Sprintf("could get available balance for %s: %s", stepState.Ref, err))
784803
}
785-
// TODO: https://smartcontract-it.atlassian.net/browse/CRE-461 if availability is math.MaxInt64 there is no limit. Possibly flag this in a different way.
786-
err = meteringReport.Deduct(stepState.Ref, availableForCall)
787-
if err != nil {
788-
l.Error(fmt.Sprintf("could not deduct balance for capability request %s: %s", stepState.Ref, err))
804+
805+
if spendLimit.Valid {
806+
err = meteringReport.Deduct(stepState.Ref, spendLimit.Decimal)
807+
if err != nil {
808+
l.Error(fmt.Sprintf("could not deduct balance for capability request %s: %s", stepState.Ref, err))
809+
}
810+
811+
spendLimits = meteringReport.CreditToSpendingLimits(info, spendLimit.Decimal)
789812
}
790813
} else {
791814
e.metrics.With(platform.KeyWorkflowID, e.workflow.id).IncrementWorkflowMissingMeteringReport(ctx)
@@ -799,16 +822,9 @@ func (e *Engine) workerForStepRequest(ctx context.Context, msg stepRequest) {
799822
// TODO: https://smartcontract-it.atlassian.net/browse/CRE-461
800823
// convert balance to CapabilityInfo resource types for use in Capability call
801824
// pass deducted amount as max spend to capability.Execute
802-
inputs, response, sErr := e.executeStep(ctx, l, msg)
825+
inputs, response, sErr := e.executeStep(ctx, l, msg, spendLimits)
803826
stepExecutionDuration := time.Since(stepExecutionStartTime).Seconds()
804827

805-
curStepID := "UNSET"
806-
curStep, verr := e.workflow.Vertex(msg.stepRef)
807-
if verr == nil {
808-
curStepID = curStep.ID
809-
} else {
810-
l.Errorf("failed to resolve step in workflow; error %v", verr)
811-
}
812828
e.metrics.With(platform.KeyCapabilityID, curStepID).UpdateWorkflowStepDurationHistogram(ctx, int64(stepExecutionDuration))
813829

814830
var stepStatus string
@@ -954,7 +970,12 @@ func (e *Engine) configForStep(ctx context.Context, lggr logger.Logger, step *st
954970
}
955971

956972
// executeStep executes the referenced capability within a step and returns the result.
957-
func (e *Engine) executeStep(ctx context.Context, lggr logger.Logger, msg stepRequest) (*values.Map, capabilities.CapabilityResponse, error) {
973+
func (e *Engine) executeStep(
974+
ctx context.Context,
975+
lggr logger.Logger,
976+
msg stepRequest,
977+
spendLimits []capabilities.SpendLimit,
978+
) (*values.Map, capabilities.CapabilityResponse, error) {
958979
curStep, err := e.workflow.Vertex(msg.stepRef)
959980
if err != nil {
960981
return nil, capabilities.CapabilityResponse{}, err
@@ -1011,6 +1032,7 @@ func (e *Engine) executeStep(ctx context.Context, lggr logger.Logger, msg stepRe
10111032
WorkflowDonConfigVersion: ln.WorkflowDON.ConfigVersion,
10121033
ReferenceID: msg.stepRef,
10131034
DecodedWorkflowName: e.workflow.name.String(),
1035+
SpendLimits: spendLimits,
10141036
},
10151037
}
10161038

core/services/workflows/metering/balance_store.go

Lines changed: 72 additions & 67 deletions
Original file line numberDiff line numberDiff line change
@@ -10,34 +10,27 @@ import (
1010
)
1111

1212
var (
13-
ErrInsufficientBalance = errors.New("insufficient balance")
14-
ErrInvalidAmount = errors.New("amount must be greater than 0")
13+
ErrInsufficientBalance = errors.New("insufficient balance")
14+
ErrInvalidAmount = errors.New("amount must be greater than 0")
15+
ErrResourceTypeNotFound = errors.New("could not find conversion rate, continuing as 1:1")
1516
)
1617

18+
// balanceStore is a locked down interface to the in-execution credit balance.
19+
// no state change details (like switching to metering mode) should be handled in it;
20+
// rather consumers should consider errors core to business logic of metering/billing.
1721
type balanceStore struct {
18-
// Whether negative balances should return an error
19-
allowNegative bool
2022
// A balance of credits
21-
balance int64
23+
balance decimal.Decimal
2224
// Conversion rates of resource dimensions to number of units per credit
2325
conversions map[string]decimal.Decimal // TODO flip this
24-
lggr logger.Logger
2526
mu sync.RWMutex
2627
}
2728

28-
type BalanceStore interface {
29-
Get() (balance int64)
30-
GetAs(unit string) (balance int64)
31-
Minus(amount int64) error
32-
MinusAs(unit string, amount int64) error
33-
Add(amount int64) error
34-
AddAs(unit string, amount int64) error
35-
AllowNegative()
36-
}
37-
38-
var _ BalanceStore = (BalanceStore)(nil)
39-
40-
func NewBalanceStore(startingBalance int64, conversions map[string]decimal.Decimal, lggr logger.Logger) *balanceStore {
29+
func NewBalanceStore(
30+
startingBalance decimal.Decimal,
31+
conversions map[string]decimal.Decimal,
32+
lggr logger.Logger,
33+
) *balanceStore {
4134
// validations
4235
for resource, rate := range conversions {
4336
if rate.IsNegative() {
@@ -48,123 +41,135 @@ func NewBalanceStore(startingBalance int64, conversions map[string]decimal.Decim
4841
}
4942

5043
return &balanceStore{
51-
allowNegative: false,
52-
balance: startingBalance,
53-
conversions: conversions,
54-
lggr: lggr,
44+
balance: startingBalance,
45+
conversions: conversions,
5546
}
5647
}
5748

5849
// convertToBalance converts a resource dimension amount to a credit amount.
5950
// This method should only be used under a read lock.
60-
func (bs *balanceStore) convertToBalance(fromUnit string, amount int64) (credits int64) {
61-
rate, ok := bs.conversions[fromUnit]
51+
func (bs *balanceStore) convertToBalance(fromResourceType string, amount decimal.Decimal) (decimal.Decimal, error) {
52+
rate, ok := bs.conversions[fromResourceType]
6253
if !ok {
63-
// Fail open, continue optimistically
64-
bs.lggr.Errorw("could not find conversion rate, continuing as 1:1", "unit", fromUnit)
65-
rate = decimal.NewFromInt(1)
54+
return amount, ErrResourceTypeNotFound
6655
}
67-
return decimal.NewFromInt(amount).Mul(rate).RoundUp(0).IntPart()
56+
57+
return amount.Mul(rate), nil
6858
}
6959

7060
// ConvertToBalance converts a resource dimensions amount to a credit amount.
71-
func (bs *balanceStore) ConvertToBalance(fromUnit string, amount int64) (credits int64) {
61+
func (bs *balanceStore) ConvertToBalance(fromResourceType string, amount decimal.Decimal) (decimal.Decimal, error) {
7262
bs.mu.RLock()
7363
defer bs.mu.RUnlock()
74-
return bs.convertToBalance(fromUnit, amount)
64+
65+
return bs.convertToBalance(fromResourceType, amount)
7566
}
7667

7768
// convertFromBalance converts a credit amount to a resource dimensions amount.
7869
// This method should only be used under a read lock.
79-
func (bs *balanceStore) convertFromBalance(toUnit string, amount int64) (resources int64) {
80-
rate, ok := bs.conversions[toUnit]
70+
func (bs *balanceStore) convertFromBalance(toResourceType string, amount decimal.Decimal) (decimal.Decimal, error) {
71+
rate, ok := bs.conversions[toResourceType]
8172
if !ok {
82-
// Fail open, continue optimistically
83-
bs.lggr.Errorw("could not find conversion rate, continuing as 1:1", "unit", toUnit)
84-
rate = decimal.NewFromInt(1)
73+
return amount, ErrResourceTypeNotFound
8574
}
86-
return decimal.NewFromInt(amount).Div(rate).RoundUp(0).IntPart()
75+
76+
return amount.Div(rate), nil
8777
}
8878

8979
// ConvertFromBalance converts a credit amount to a resource dimensions amount.
90-
func (bs *balanceStore) ConvertFromBalance(toUnit string, amount int64) (resources int64) {
80+
func (bs *balanceStore) ConvertFromBalance(toResourceType string, amount decimal.Decimal) (decimal.Decimal, error) {
9181
bs.mu.RLock()
9282
defer bs.mu.RUnlock()
93-
return bs.convertFromBalance(toUnit, amount)
83+
84+
return bs.convertFromBalance(toResourceType, amount)
9485
}
9586

9687
// Get returns the current credit balance
97-
func (bs *balanceStore) Get() (balance int64) {
88+
func (bs *balanceStore) Get() decimal.Decimal {
9889
bs.mu.RLock()
9990
defer bs.mu.RUnlock()
91+
10092
return bs.balance
10193
}
10294

10395
// GetAs returns the current universal credit balance expressed as a resource dimensions.
104-
func (bs *balanceStore) GetAs(unit string) (balance int64) {
96+
func (bs *balanceStore) GetAs(unit string) (decimal.Decimal, error) {
10597
bs.mu.RLock()
10698
defer bs.mu.RUnlock()
107-
if bs.balance <= 0 {
108-
return 0
109-
}
99+
110100
return bs.convertFromBalance(unit, bs.balance)
111101
}
112102

113103
// Minus lowers the current credit balance.
114-
func (bs *balanceStore) Minus(amount int64) error {
104+
func (bs *balanceStore) Minus(amount decimal.Decimal) error {
115105
bs.mu.Lock()
116106
defer bs.mu.Unlock()
117-
if amount <= 0 {
107+
108+
if amount.LessThan(decimal.Zero) {
118109
return ErrInvalidAmount
119110
}
120-
if amount > bs.balance && !bs.allowNegative {
111+
112+
if amount.GreaterThan(bs.balance) {
121113
return ErrInsufficientBalance
122114
}
123-
bs.balance -= amount
115+
116+
bs.balance = bs.balance.Sub(amount)
117+
124118
return nil
125119
}
126120

127121
// MinusAs lowers the current credit balance based on an amount of resource dimensions.
128-
func (bs *balanceStore) MinusAs(unit string, amount int64) error {
122+
func (bs *balanceStore) MinusAs(resourceType string, amount decimal.Decimal) error {
129123
bs.mu.Lock()
130124
defer bs.mu.Unlock()
131-
if amount <= 0 {
125+
126+
if amount.LessThan(decimal.Zero) {
132127
return ErrInvalidAmount
133128
}
134-
balToMinus := bs.convertToBalance(unit, amount)
135-
if balToMinus > bs.balance && !bs.allowNegative {
129+
130+
balToMinus, err := bs.convertToBalance(resourceType, amount)
131+
if err != nil {
132+
return err
133+
}
134+
135+
if balToMinus.GreaterThan(bs.balance) {
136136
return ErrInsufficientBalance
137137
}
138-
bs.balance -= balToMinus
138+
139+
bs.balance = bs.balance.Sub(balToMinus)
140+
139141
return nil
140142
}
141143

142144
// Add increases the current credit balance.
143-
func (bs *balanceStore) Add(amount int64) error {
145+
func (bs *balanceStore) Add(amount decimal.Decimal) error {
144146
bs.mu.Lock()
145147
defer bs.mu.Unlock()
146-
if amount <= 0 {
148+
149+
if amount.LessThan(decimal.Zero) {
147150
return ErrInvalidAmount
148151
}
149-
bs.balance += amount
152+
153+
bs.balance = bs.balance.Add(amount)
154+
150155
return nil
151156
}
152157

153158
// AddAs increases the current credit balance based on an amount of resource dimensions.
154-
func (bs *balanceStore) AddAs(unit string, amount int64) error {
159+
func (bs *balanceStore) AddAs(resourceType string, amount decimal.Decimal) error {
155160
bs.mu.Lock()
156161
defer bs.mu.Unlock()
157-
if amount <= 0 {
162+
163+
if amount.LessThan(decimal.Zero) {
158164
return ErrInvalidAmount
159165
}
160-
balToAdd := bs.convertToBalance(unit, amount)
161-
bs.balance += balToAdd
162-
return nil
163-
}
164166

165-
// AllowNegative turns on the flag to allow negative balances.
166-
func (bs *balanceStore) AllowNegative() {
167-
bs.mu.Lock()
168-
defer bs.mu.Unlock()
169-
bs.allowNegative = true
167+
bal, err := bs.convertToBalance(resourceType, amount)
168+
if err != nil {
169+
return err
170+
}
171+
172+
bs.balance = bs.balance.Add(bal)
173+
174+
return nil
170175
}

0 commit comments

Comments
 (0)