Minimum payment should be controllable on the job level

core/cmd/renderer.go

@@ -161,12 +161,13 @@ func (rt RendererTable) renderJobRun(run presenters.JobRun) error {
 }
 
 func (rt RendererTable) renderJobSingles(j presenters.JobSpec) error {
-	table := rt.newTable([]string{"ID", "Created At", "Start At", "End At"})
+	table := rt.newTable([]string{"ID", "Created At", "Start At", "End At", "Min Payment"})
 	table.Append([]string{
 		j.ID,
 		j.FriendlyCreatedAt(),
 		j.FriendlyStartAt(),
 		j.FriendlyEndAt(),
+		j.FriendlyMinPayment(),
 	})
 	render("Job", table)
 	return nil

core/services/runs.go

@@ -57,6 +57,18 @@ func ExecuteJobWithRunRequest(
 	return run, createAndTrigger(run, store)
 }
 
+// MeetsMinimumPayment is a helper that returns true if jobrun received
+// sufficient payment (more than jobspec's MinimumPayment) to be considered successful
+func MeetsMinimumPayment(
+	expectedMinJobPayment *assets.Link,
+	actualRunPayment *assets.Link) bool {
+	// input.Amount is always present for runs triggered by ethlogs
+	if actualRunPayment == nil || expectedMinJobPayment == nil || expectedMinJobPayment.IsZero() {
+		return true
+	}
+	return expectedMinJobPayment.Cmp(actualRunPayment) < 1
+}
+
 // NewRun returns a run from an input job, in an initial state ready for
 // processing by the job runner system
 func NewRun(
@@ -86,6 +98,22 @@ func NewRun(
 	run.CreationHeight = models.NewBig(currentHeight)
 	run.ObservedHeight = models.NewBig(currentHeight)
 
+	if !MeetsMinimumPayment(job.MinPayment, input.Amount) {
+		logger.Infow("Rejecting run with insufficient payment", []interface{}{
+			"run", run.ID,
+			"job", run.JobSpecID,
+			"input_amount", input.Amount,
+			"required_amount", job.MinPayment,
+		}...)
+
+		err := fmt.Errorf(
+			"Rejecting job %s with payment %s below job-specific-minimum threshold (%s)",
+			job.ID,
+			input.Amount,
+			job.MinPayment.Text(10))
+		run.SetError(err)
+	}
+
 	cost := assets.NewLink(0)
 	for i, taskRun := range run.TaskRuns {
 		adapter, err := adapters.For(taskRun.TaskSpec, store)

core/services/runs_test.go

@@ -51,6 +51,29 @@ func TestNewRun(t *testing.T) {
 	assert.False(t, run.TaskRuns[0].Confirmations.Valid)
 }
 
+func TestNewRun_MeetsMinimumPayment(t *testing.T) {
+	tests := []struct {
+		name            string
+		MinJobPayment   *assets.Link
+		RunPayment      *assets.Link
+		meetsMinPayment bool
+	}{
+		{"insufficient payment", assets.NewLink(100), assets.NewLink(10), false},
+		{"sufficient payment (strictly greater)", assets.NewLink(1), assets.NewLink(10), true},
+		{"sufficient payment (equal)", assets.NewLink(10), assets.NewLink(10), true},
+		{"runs that do not accept payments must return true", assets.NewLink(10), nil, true},
+		{"return true when minpayment is not specified in jobspec", nil, assets.NewLink(0), true},
+	}
+
+	for _, tt := range tests {
+		test := tt
+		t.Run(test.name, func(t *testing.T) {
+			actual := services.MeetsMinimumPayment(test.MinJobPayment, test.RunPayment)
+			assert.Equal(t, test.meetsMinPayment, actual)
+		})
+	}
+}
+
 func TestNewRun_requiredPayment(t *testing.T) {
 	store, cleanup := cltest.NewStore(t)
 	defer cleanup()
@@ -62,22 +85,25 @@ func TestNewRun_requiredPayment(t *testing.T) {
 	require.NoError(t, store.CreateBridgeType(bt))
 
 	tests := []struct {
-		name           string
-		payment        *assets.Link
-		minimumPayment *assets.Link
-		expectedStatus models.RunStatus
+		name                  string
+		payment               *assets.Link
+		minimumConfigPayment  *assets.Link
+		minimumJobSpecPayment *assets.Link
+		expectedStatus        models.RunStatus
 	}{
-		{"creates runnable job", nil, assets.NewLink(0), models.RunStatusInProgress},
-		{"insufficient payment as specified by config", assets.NewLink(9), assets.NewLink(10), models.RunStatusErrored},
-		{"sufficient payment as specified by config", assets.NewLink(10), assets.NewLink(10), models.RunStatusInProgress},
-		{"insufficient payment as specified by adapter", assets.NewLink(9), assets.NewLink(0), models.RunStatusErrored},
-		{"sufficient payment as specified by adapter", assets.NewLink(10), assets.NewLink(0), models.RunStatusInProgress},
+		{"creates runnable job", nil, assets.NewLink(0), assets.NewLink(0), models.RunStatusInProgress},
+		{"insufficient payment as specified by config", assets.NewLink(9), assets.NewLink(10), assets.NewLink(0), models.RunStatusErrored},
+		{"sufficient payment as specified by config", assets.NewLink(10), assets.NewLink(10), assets.NewLink(0), models.RunStatusInProgress},
+		{"insufficient payment as specified by adapter", assets.NewLink(9), assets.NewLink(0), assets.NewLink(0), models.RunStatusErrored},
+		{"sufficient payment as specified by adapter", assets.NewLink(10), assets.NewLink(0), assets.NewLink(0), models.RunStatusInProgress},
+		{"insufficient payment as specified by jobSpec MinPayment", assets.NewLink(9), assets.NewLink(0), assets.NewLink(10), models.RunStatusErrored},
+		{"sufficient payment as specified by jobSpec MinPayment", assets.NewLink(10), assets.NewLink(0), assets.NewLink(10), models.RunStatusInProgress},
 	}
 
 	for _, tt := range tests {
 		test := tt
 		t.Run(test.name, func(t *testing.T) {
-			store.Config.Set("MINIMUM_CONTRACT_PAYMENT", test.minimumPayment)
+			store.Config.Set("MINIMUM_CONTRACT_PAYMENT", test.minimumConfigPayment)
 
 			jobSpec := models.NewJob()
 			jobSpec.Tasks = []models.TaskSpec{{
@@ -86,6 +112,7 @@ func TestNewRun_requiredPayment(t *testing.T) {
 			jobSpec.Initiators = []models.Initiator{{
 				Type: models.InitiatorEthLog,
 			}}
+			jobSpec.MinPayment = test.minimumJobSpecPayment
 
 			inputResult := models.RunResult{Data: input, Amount: test.payment}
 

core/store/migrations/migrate.go

@@ -14,6 +14,7 @@ import (
 	"github.com/smartcontractkit/chainlink/core/store/migrations/migration1560881855"
 	"github.com/smartcontractkit/chainlink/core/store/migrations/migration1560886530"
 	"github.com/smartcontractkit/chainlink/core/store/migrations/migration1560924400"
+	"github.com/smartcontractkit/chainlink/core/store/migrations/migration1565139192"
 	gormigrate "gopkg.in/gormigrate.v1"
 )
 
@@ -60,6 +61,10 @@ func Migrate(db *gorm.DB) error {
 			ID:      "1560881855",
 			Migrate: migration1560881855.Migrate,
 		},
+		{
+			ID:      "1565139192",
+			Migrate: migration1565139192.Migrate,
+		},
 	}
 
 	m := gormigrate.New(db, &options, migrations)

core/store/migrations/migrate_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/smartcontractkit/chainlink/core/store/migrations/migration1560881846"
 	"github.com/smartcontractkit/chainlink/core/store/migrations/migration1560881855"
 	"github.com/smartcontractkit/chainlink/core/store/migrations/migration1560886530"
+	"github.com/smartcontractkit/chainlink/core/store/migrations/migration1565139192"
 	"github.com/smartcontractkit/chainlink/core/store/models"
 	"github.com/smartcontractkit/chainlink/core/store/orm"
 	"github.com/stretchr/testify/assert"
@@ -121,6 +122,7 @@ func TestMigrate_Migration1560881855(t *testing.T) {
 	require.NoError(t, migration1560881846.Migrate(db))
 	require.NoError(t, migration1560886530.Migrate(db))
 	require.NoError(t, migration1560924400.Migrate(db))
+	require.NoError(t, migration1565139192.Migrate(db))
 
 	j := models.NewJob()
 	i := models.Initiator{Type: models.InitiatorWeb}
@@ -177,6 +179,28 @@ func TestMigrate_Migration1560881846(t *testing.T) {
 	assert.Equal(t, int64(8616460799), headFound.Number)
 }
 
+func TestMigrate_Migration1565139192(t *testing.T) {
+	orm, cleanup := bootstrapORM(t)
+	defer cleanup()
+	db := orm.DB
+
+	require.NoError(t, migration0.Migrate(db))
+	require.NoError(t, migration1565139192.Migrate(db))
+	specNoPayment := models.NewJobFromRequest(models.JobSpecRequest{})
+	specWithPayment := models.NewJobFromRequest(models.JobSpecRequest{
+		MinPayment: assets.NewLink(5),
+	})
+	specOneFound := models.JobSpec{}
+	specTwoFound := models.JobSpec{}
+
+	require.NoError(t, db.Create(&specWithPayment).Error)
+	require.NoError(t, db.Create(&specNoPayment).Error)
+	require.NoError(t, db.Where("id = ?", specNoPayment.ID).Find(&specOneFound).Error)
+	require.Equal(t, assets.NewLink(0), specNoPayment.MinPayment)
+	require.NoError(t, db.Where("id = ?", specWithPayment.ID).Find(&specTwoFound).Error)
+	require.Equal(t, assets.NewLink(5), specWithPayment.MinPayment)
+}
+
 func TestMigrate_NewerVersionGuard(t *testing.T) {
 	orm, cleanup := bootstrapORM(t)
 	defer cleanup()

core/store/migrations/migration0/migrate.go

@@ -7,6 +7,7 @@ import (
 	"github.com/jinzhu/gorm"
 	"github.com/pkg/errors"
 	"github.com/smartcontractkit/chainlink/core/store/models"
+	"gopkg.in/guregu/null.v3"
 )
 
 func Migrate(tx *gorm.DB) error {
@@ -22,7 +23,7 @@ func Migrate(tx *gorm.DB) error {
 	if err := tx.AutoMigrate(&Head{}).Error; err != nil {
 		return errors.Wrap(err, "failed to auto migrate Head")
 	}
-	if err := tx.AutoMigrate(&models.JobSpec{}).Error; err != nil {
+	if err := tx.AutoMigrate(JobSpec{}).Error; err != nil {
 		return errors.Wrap(err, "failed to auto migrate JobSpec")
 	}
 	if err := tx.AutoMigrate(&models.Initiator{}).Error; err != nil {
@@ -120,3 +121,14 @@ type RunRequest struct {
 	Requester *common.Address
 	CreatedAt time.Time
 }
+
+// JobSpec is a capture of the model representing Head before migration1565139192
+type JobSpec struct {
+	ID         string             `json:"id,omitempty" gorm:"primary_key;not null"`
+	CreatedAt  time.Time          `json:"createdAt" gorm:"index"`
+	Initiators []models.Initiator `json:"initiators"`
+	Tasks      []models.TaskSpec  `json:"tasks"`
+	StartAt    null.Time          `json:"startAt" gorm:"index"`
+	EndAt      null.Time          `json:"endAt" gorm:"index"`
+	DeletedAt  null.Time          `json:"-" gorm:"index"`
+}

core/store/migrations/migration1565139192/migrate.go

@@ -0,0 +1,13 @@
+package migration1565139192
+
+import (
+	"github.com/jinzhu/gorm"
+	"github.com/pkg/errors"
+)
+
+func Migrate(tx *gorm.DB) error {
+	if err := tx.Exec(`ALTER TABLE job_specs ADD min_payment varchar(255)`).Error; err != nil {
+		return errors.Wrap(err, "failed to add MinPayment to JobSpec")
+	}
+	return nil
+}

core/store/models/job_spec.go

@@ -22,6 +22,7 @@ type JobSpecRequest struct {
 	Tasks      []TaskSpecRequest  `json:"tasks"`
 	StartAt    null.Time          `json:"startAt"`
 	EndAt      null.Time          `json:"endAt"`
+	MinPayment *assets.Link       `json:"minPayment"`
 }
 
 // InitiatorRequest represents a schema for incoming initiator requests as used by the API.
@@ -41,13 +42,14 @@ type TaskSpecRequest struct {
 // for a given contract. It contains the Initiators, Tasks (which are the
 // individual steps to be carried out), StartAt, EndAt, and CreatedAt fields.
 type JobSpec struct {
-	ID         string      `json:"id,omitempty" gorm:"primary_key;not null"`
-	CreatedAt  time.Time   `json:"createdAt" gorm:"index"`
-	Initiators []Initiator `json:"initiators"`
-	Tasks      []TaskSpec  `json:"tasks"`
-	StartAt    null.Time   `json:"startAt" gorm:"index"`
-	EndAt      null.Time   `json:"endAt" gorm:"index"`
-	DeletedAt  null.Time   `json:"-" gorm:"index"`
+	ID         string       `json:"id,omitempty" gorm:"primary_key;not null"`
+	CreatedAt  time.Time    `json:"createdAt" gorm:"index"`
+	Initiators []Initiator  `json:"initiators"`
+	MinPayment *assets.Link `json:"minPayment" gorm:"type:varchar(255)"`
+	Tasks      []TaskSpec   `json:"tasks"`
+	StartAt    null.Time    `json:"startAt" gorm:"index"`
+	EndAt      null.Time    `json:"endAt" gorm:"index"`
+	DeletedAt  null.Time    `json:"-" gorm:"index"`
 }
 
 // GetID returns the ID of this structure for jsonapi serialization.
@@ -70,8 +72,9 @@ func (j *JobSpec) SetID(value string) error {
 // the CreatedAt field to the time of invokation.
 func NewJob() JobSpec {
 	return JobSpec{
-		ID:        utils.NewBytes32ID(),
-		CreatedAt: time.Now(),
+		ID:         utils.NewBytes32ID(),
+		CreatedAt:  time.Now(),
+		MinPayment: assets.NewLink(0),
 	}
 }
 
@@ -100,6 +103,9 @@ func NewJobFromRequest(jsr JobSpecRequest) JobSpec {
 
 	jobSpec.EndAt = jsr.EndAt
 	jobSpec.StartAt = jsr.StartAt
+	if jsr.MinPayment != nil {
+		jobSpec.MinPayment = jsr.MinPayment
+	}
 	return jobSpec
 }
 

core/store/models/job_spec_test.go

@@ -22,10 +22,11 @@ func TestNewJobFromRequest(t *testing.T) {
 	require.NoError(t, store.CreateJob(&j1))
 
 	jsr := models.JobSpecRequest{
-		Initiators: cltest.BuildInitiatorRequests(t, j1.Initiators),
-		Tasks:      cltest.BuildTaskRequests(t, j1.Tasks),
-		StartAt:    j1.StartAt,
-		EndAt:      j1.EndAt,
+		Initiators:        cltest.BuildInitiatorRequests(t, j1.Initiators),
+		Tasks:             cltest.BuildTaskRequests(t, j1.Tasks),
+		StartAt:           j1.StartAt,
+		EndAt:             j1.EndAt,
+		MinPayment: assets.NewLink(5),
 	}
 
 	j2 := models.NewJobFromRequest(jsr)
@@ -35,11 +36,13 @@ func TestNewJobFromRequest(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Len(t, fetched1.Initiators, 1)
 	assert.Len(t, fetched1.Tasks, 1)
+	assert.Equal(t, fetched1.MinPayment, assets.NewLink(0))
 
 	fetched2, err := store.FindJob(j2.ID)
 	assert.NoError(t, err)
 	assert.Len(t, fetched2.Initiators, 1)
 	assert.Len(t, fetched2.Tasks, 1)
+	assert.Equal(t, fetched2.MinPayment, assets.NewLink(5))
 }
 
 func TestJobSpec_Save(t *testing.T) {

core/store/presenters/presenters.go

@@ -289,6 +289,15 @@ func (job JobSpec) FriendlyEndAt() string {
 	return ""
 }
 
+// FriendlyMinPayment returns a formatted string of the Job's
+// Minimum Link Payment threshold
+func (job JobSpec) FriendlyMinPayment() string {
+	if job.MinPayment == nil {
+		return assets.NewLink(0).Text(10)
+	}
+	return job.MinPayment.Text(10)
+}
+
 // FriendlyInitiators returns the list of Initiator types as
 // a comma separated string.
 func (job JobSpec) FriendlyInitiators() string {

operator_ui/@types/db/index.d.ts

@@ -51,6 +51,7 @@ export interface JobRun {
 
 export interface JobSpec {
     id: string
+    minPayment: number
     createdAt: string
     startAt: Date | null
     endAt: Date | null

operator_ui/__tests__/containers/Jobs/Show.test.js

@@ -8,7 +8,7 @@ import { MemoryRouter } from 'react-router-dom'
 import { ConnectedShow as Show } from 'containers/Jobs/Show'
 import isoDate, { MINUTE_MS } from 'test-helpers/isoDate'
 import jsonApiJobSpecRunsFactory from 'factories/jsonApiJobSpecRuns'
-import { GWEI_PER_TOKEN } from '../../../src/utils/constants'
+import { GWEI_PER_TOKEN, WEI_PER_TOKEN } from '../../../src/utils/constants'
 
 const mountShow = props =>
   mountWithTheme(
@@ -23,14 +23,15 @@ describe('containers/Jobs/Show', () => {
   const jobSpecId = 'c60b9927eeae43168ddbe92584937b1b'
   const jobRunId = 'ad24b72c12f441b99b9877bcf6cb506e'
   it('renders the details of the job spec, its latest runs, its task list entries and its total earnings', async () => {
-    expect.assertions(8)
+    expect.assertions(9)
 
     const minuteAgo = isoDate(Date.now() - MINUTE_MS)
     const jobSpecResponse = jsonApiJobSpecFactory({
       id: jobSpecId,
       initiators: [{ type: 'web' }],
       createdAt: minuteAgo,
-      earnings: GWEI_PER_TOKEN
+      earnings: GWEI_PER_TOKEN,
+      minPayment: 100 * WEI_PER_TOKEN
     })
     global.fetch.getOnce(`/v2/specs/${jobSpecId}`, jobSpecResponse)
 
@@ -56,6 +57,7 @@ describe('containers/Jobs/Show', () => {
     expect(wrapper.text()).toContain('1.000000')
     expect(wrapper.text()).toContain('Httpget')
     expect(wrapper.text()).toContain('Run Count1')
+    expect(wrapper.text()).toContain('Minimum Payment100 Link')
     expect(wrapper.text()).toContain('Pending')
     expect(wrapper.text()).not.toContain('View More')
   })

operator_ui/src/containers/Jobs/Show.tsx

@@ -25,11 +25,13 @@ import { GWEI_PER_TOKEN } from '../../utils/constants'
 import { formatInitiators } from '../../utils/jobSpecInitiators'
 import matchRouteAndMapDispatchToProps from '../../utils/matchRouteAndMapDispatchToProps'
 import RegionalNav from './RegionalNav'
+import formatMinPayment from '../../utils/formatWeiAsset'
 
 const renderJobSpec = (job: IJobSpec, recentRunsCount: number) => {
   const info = {
     runCount: recentRunsCount,
-    initiator: formatInitiators(job.initiators)
+    initiator: formatInitiators(job.initiators),
+    minimumPayment: `${formatMinPayment(job.minPayment) || 0} Link`
   }
 
   return (