Skip to content

Commit 232a6ad

Browse files
jkongiejkongie
authored
feat(feeds): remove stream ID lookup when approving data streams jobs (#22508)
Match data streams (job.Stream) specs only by external job ID when approving job proposals, the same as CRESettings and CCV job types. Previously, a stream spec with a different external job ID but the same stream ID as a running job would trigger the force-replace path (delete + recreate). That auto-delete behavior is now removed; stream job uniqueness is determined exclusively by external job ID. OPDATA-5937
1 parent 7ce383f commit 232a6ad

4 files changed

Lines changed: 51 additions & 251 deletions

File tree

‎core/services/feeds/service.go‎

Lines changed: 6 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -1052,29 +1052,19 @@ func (s *service) ApproveSpec(ctx context.Context, id int64, force bool) error {
10521052
if txerr != nil && !errors.Is(txerr, sql.ErrNoRows) {
10531053
return fmt.Errorf("failed while checking for existing ccip job: %w", txerr)
10541054
}
1055-
case job.StandardCapabilities:
1056-
// Only possible to match standard capabilities by external job id
1057-
// no-op
10581055
case job.Gateway:
10591056
existingJobID, txerr = tx.jobORM.FindGatewayJobID(ctx, *j.GatewaySpec)
10601057
// Return an error if the repository errors. If there is a not found
10611058
// error we want to continue with approving the job.
10621059
if txerr != nil && !errors.Is(txerr, sql.ErrNoRows) {
10631060
return fmt.Errorf("failed while checking for existing gateway job: %w", txerr)
10641061
}
1065-
case job.Stream:
1066-
existingJobID, txerr = tx.jobORM.FindJobIDByStreamID(ctx, *j.StreamID)
1067-
// Return an error if the repository errors. If there is a not found
1068-
// error we want to continue with approving the job.
1069-
if txerr != nil && !errors.Is(txerr, sql.ErrNoRows) {
1070-
return fmt.Errorf("failed while checking for existing stream job: %w", txerr)
1071-
}
1072-
case job.CRESettings:
1073-
// Only possible to match CRE Setting by external job id
1074-
// no-op
1075-
case job.CCVCommitteeVerifier, job.CCVExecutor:
1076-
// Only possible to match CCV jobs by external job id
1077-
// no-op
1062+
case job.CRESettings,
1063+
job.Stream,
1064+
job.CCVCommitteeVerifier,
1065+
job.CCVExecutor,
1066+
job.StandardCapabilities:
1067+
// NOOP: These jobs are only matched by external job ID, so do nothing
10781068
default:
10791069
return errors.Errorf("unsupported job type when approving job proposal specs: %s", j.Type)
10801070
}

‎core/services/feeds/service_test.go‎

Lines changed: 39 additions & 156 deletions
Original file line numberDiff line numberDiff line change
@@ -3874,10 +3874,6 @@ func Test_Service_ApproveSpec_Stream(t *testing.T) {
38743874
Version: 1,
38753875
Definition: fmt.Sprintf(StreamTestSpecTemplate, streamName, externalJobID.String(), streamID),
38763876
}
3877-
j = job.Job{
3878-
ID: 1,
3879-
ExternalJobID: externalJobID,
3880-
}
38813877
)
38823878

38833879
testCases := []struct {
@@ -3897,7 +3893,6 @@ func Test_Service_ApproveSpec_Stream(t *testing.T) {
38973893
svc.connMgr.On("GetClient", jp.FeedsManagerID).Return(svc.fmsClient, nil)
38983894
svc.jobORM.On("AssertBridgesExist", mock.Anything, mock.IsType(pipeline.Pipeline{})).Return(nil)
38993895
svc.jobORM.On("FindJobByExternalJobID", mock.Anything, externalJobID).Return(job.Job{}, sql.ErrNoRows)
3900-
svc.jobORM.On("FindJobIDByStreamID", mock.Anything, mock.Anything).Return(int32(0), sql.ErrNoRows)
39013896

39023897
svc.spawner.
39033898
On("CreateJob",
@@ -3929,25 +3924,17 @@ func Test_Service_ApproveSpec_Stream(t *testing.T) {
39293924
force: false,
39303925
},
39313926
{
3932-
name: "cancelled spec success when no other spec is approved",
3927+
// Stream jobs are matched only by external job ID; a collision
3928+
// on stream ID alone must not block approval at the feeds layer.
3929+
// The strict mock asserts FindJobIDByStreamID is never called.
3930+
name: "pending job success when another job has the same stream ID",
39333931
httpTimeout: commonconfig.MustNewDuration(1 * time.Minute),
39343932
before: func(svc *TestService) {
3935-
otherSpec := feeds.JobProposalSpec{
3936-
ID: 21,
3937-
Status: feeds.SpecStatusRevoked,
3938-
JobProposalID: jp.ID,
3939-
Version: 2,
3940-
Definition: fmt.Sprintf(StreamTestSpecTemplate, streamName, externalJobID.String(), streamID),
3941-
}
3942-
3943-
svc.connMgr.On("GetClient", jp.FeedsManagerID).Return(svc.fmsClient, nil)
3944-
svc.orm.On("GetSpec", mock.Anything, cancelledSpec.ID, mock.Anything).Return(cancelledSpec, nil)
3933+
svc.orm.On("GetSpec", mock.Anything, spec.ID).Return(spec, nil)
39453934
svc.orm.On("GetJobProposal", mock.Anything, jp.ID).Return(jp, nil)
3946-
svc.orm.On("ListSpecsByJobProposalIDs", mock.Anything, []int64{cancelledSpec.JobProposalID}).
3947-
Return([]feeds.JobProposalSpec{otherSpec, *cancelledSpec}, nil)
3935+
svc.connMgr.On("GetClient", jp.FeedsManagerID).Return(svc.fmsClient, nil)
39483936
svc.jobORM.On("AssertBridgesExist", mock.Anything, mock.IsType(pipeline.Pipeline{})).Return(nil)
39493937
svc.jobORM.On("FindJobByExternalJobID", mock.Anything, externalJobID).Return(job.Job{}, sql.ErrNoRows)
3950-
svc.jobORM.On("FindJobIDByStreamID", mock.Anything, mock.Anything).Return(int32(0), sql.ErrNoRows)
39513938

39523939
svc.spawner.
39533940
On("CreateJob",
@@ -3961,7 +3948,7 @@ func Test_Service_ApproveSpec_Stream(t *testing.T) {
39613948
Return(nil)
39623949
svc.orm.On("ApproveSpec",
39633950
mock.Anything,
3964-
cancelledSpec.ID,
3951+
spec.ID,
39653952
externalJobID,
39663953
).Return(nil)
39673954
svc.fmsClient.On("ApprovedJob",
@@ -3975,68 +3962,28 @@ func Test_Service_ApproveSpec_Stream(t *testing.T) {
39753962
svc.orm.On("WithDataSource", mock.Anything).Return(feeds.ORM(svc.orm))
39763963
svc.jobORM.On("WithDataSource", mock.Anything).Return(job.ORM(svc.jobORM))
39773964
},
3978-
id: cancelledSpec.ID,
3965+
id: spec.ID,
39793966
force: false,
39803967
},
39813968
{
3982-
name: "cancelled spec failed when another spec is approved",
3969+
name: "cancelled spec success when no other spec is approved",
3970+
httpTimeout: commonconfig.MustNewDuration(1 * time.Minute),
39833971
before: func(svc *TestService) {
39843972
otherSpec := feeds.JobProposalSpec{
39853973
ID: 21,
3986-
Status: feeds.SpecStatusApproved,
3974+
Status: feeds.SpecStatusRevoked,
39873975
JobProposalID: jp.ID,
39883976
Version: 2,
39893977
Definition: fmt.Sprintf(StreamTestSpecTemplate, streamName, externalJobID.String(), streamID),
39903978
}
39913979

3992-
svc.orm.On("GetSpec", mock.Anything, cancelledSpec.ID, mock.Anything).Return(cancelledSpec, nil)
3980+
svc.connMgr.On("GetClient", jp.FeedsManagerID).Return(svc.fmsClient, nil)
3981+
svc.orm.On("GetSpec", mock.Anything, cancelledSpec.ID).Return(cancelledSpec, nil)
39933982
svc.orm.On("GetJobProposal", mock.Anything, jp.ID).Return(jp, nil)
39943983
svc.orm.On("ListSpecsByJobProposalIDs", mock.Anything, []int64{cancelledSpec.JobProposalID}).
39953984
Return([]feeds.JobProposalSpec{otherSpec, *cancelledSpec}, nil)
3996-
},
3997-
id: cancelledSpec.ID,
3998-
force: false,
3999-
wantErr: "the job spec with version 2 is already approved",
4000-
},
4001-
{
4002-
name: "rejected spec failed cannot be approved",
4003-
before: func(svc *TestService) {
4004-
svc.orm.On("GetSpec", mock.Anything, cancelledSpec.ID, mock.Anything).Return(rejectedSpec, nil)
4005-
svc.orm.On("GetJobProposal", mock.Anything, jp.ID).Return(jp, nil)
4006-
},
4007-
id: rejectedSpec.ID,
4008-
force: false,
4009-
wantErr: "cannot approve a rejected spec",
4010-
},
4011-
{
4012-
name: "already existing job replacement error",
4013-
httpTimeout: commonconfig.MustNewDuration(1 * time.Minute),
4014-
before: func(svc *TestService) {
4015-
svc.connMgr.On("GetClient", jp.FeedsManagerID).Return(svc.fmsClient, nil)
4016-
svc.orm.On("GetSpec", mock.Anything, spec.ID).Return(spec, nil)
4017-
svc.orm.On("GetJobProposal", mock.Anything, jp.ID).Return(jp, nil)
40183985
svc.jobORM.On("AssertBridgesExist", mock.Anything, mock.IsType(pipeline.Pipeline{})).Return(nil)
40193986
svc.jobORM.On("FindJobByExternalJobID", mock.Anything, externalJobID).Return(job.Job{}, sql.ErrNoRows)
4020-
svc.jobORM.On("FindJobIDByStreamID", mock.Anything, mock.Anything).Return(j.ID, nil)
4021-
svc.orm.On("WithDataSource", mock.Anything).Return(feeds.ORM(svc.orm))
4022-
svc.jobORM.On("WithDataSource", mock.Anything).Return(job.ORM(svc.jobORM))
4023-
},
4024-
id: spec.ID,
4025-
force: false,
4026-
wantErr: "could not approve job proposal: a job for this contract address already exists - please use the 'force' option to replace it",
4027-
},
4028-
{
4029-
name: "already existing self managed job replacement success if forced without feedID",
4030-
httpTimeout: commonconfig.MustNewDuration(1 * time.Minute),
4031-
before: func(svc *TestService) {
4032-
svc.connMgr.On("GetClient", jp.FeedsManagerID).Return(svc.fmsClient, nil)
4033-
svc.orm.On("GetSpec", mock.Anything, spec.ID).Return(spec, nil)
4034-
svc.orm.On("GetJobProposal", mock.Anything, jp.ID).Return(jp, nil)
4035-
svc.jobORM.On("AssertBridgesExist", mock.Anything, mock.IsType(pipeline.Pipeline{})).Return(nil)
4036-
svc.orm.EXPECT().GetApprovedSpec(mock.Anything, jp.ID).Return(nil, sql.ErrNoRows)
4037-
svc.jobORM.On("FindJobByExternalJobID", mock.Anything, externalJobID).Return(job.Job{}, sql.ErrNoRows)
4038-
svc.jobORM.On("FindJobIDByStreamID", mock.Anything, mock.Anything).Return(j.ID, nil)
4039-
svc.spawner.On("DeleteJob", mock.Anything, mock.Anything, j.ID).Return(nil)
40403987

40413988
svc.spawner.
40423989
On("CreateJob",
@@ -4050,7 +3997,7 @@ func Test_Service_ApproveSpec_Stream(t *testing.T) {
40503997
Return(nil)
40513998
svc.orm.On("ApproveSpec",
40523999
mock.Anything,
4053-
spec.ID,
4000+
cancelledSpec.ID,
40544001
externalJobID,
40554002
).Return(nil)
40564003
svc.fmsClient.On("ApprovedJob",
@@ -4064,99 +4011,38 @@ func Test_Service_ApproveSpec_Stream(t *testing.T) {
40644011
svc.orm.On("WithDataSource", mock.Anything).Return(feeds.ORM(svc.orm))
40654012
svc.jobORM.On("WithDataSource", mock.Anything).Return(job.ORM(svc.jobORM))
40664013
},
4067-
id: spec.ID,
4068-
force: true,
4014+
id: cancelledSpec.ID,
4015+
force: false,
40694016
},
40704017
{
4071-
name: "already existing self managed job replacement success if forced with feedID",
4072-
httpTimeout: commonconfig.MustNewDuration(1 * time.Minute),
4018+
name: "cancelled spec failed when another spec is approved",
40734019
before: func(svc *TestService) {
4074-
svc.connMgr.On("GetClient", jp.FeedsManagerID).Return(svc.fmsClient, nil)
4075-
svc.orm.On("GetSpec", mock.Anything, spec.ID).Return(&feeds.JobProposalSpec{
4076-
ID: 20,
4077-
Status: feeds.SpecStatusPending,
4020+
otherSpec := feeds.JobProposalSpec{
4021+
ID: 21,
4022+
Status: feeds.SpecStatusApproved,
40784023
JobProposalID: jp.ID,
4079-
Version: 1,
4024+
Version: 2,
40804025
Definition: fmt.Sprintf(StreamTestSpecTemplate, streamName, externalJobID.String(), streamID),
4081-
}, nil)
4082-
svc.orm.On("GetJobProposal", mock.Anything, jp.ID).Return(jp, nil)
4083-
svc.jobORM.On("AssertBridgesExist", mock.Anything, mock.IsType(pipeline.Pipeline{})).Return(nil)
4084-
svc.orm.EXPECT().GetApprovedSpec(mock.Anything, jp.ID).Return(nil, sql.ErrNoRows)
4085-
svc.jobORM.On("FindJobByExternalJobID", mock.Anything, externalJobID).Return(job.Job{}, sql.ErrNoRows)
4086-
svc.jobORM.On("FindJobIDByStreamID", mock.Anything, mock.Anything).Return(j.ID, nil)
4087-
svc.spawner.On("DeleteJob", mock.Anything, mock.Anything, j.ID).Return(nil)
4026+
}
40884027

4089-
svc.spawner.
4090-
On("CreateJob",
4091-
mock.Anything,
4092-
mock.Anything,
4093-
mock.MatchedBy(func(j *job.Job) bool {
4094-
return j.Name.String == streamName
4095-
}),
4096-
).
4097-
Run(func(args mock.Arguments) { (args.Get(2).(*job.Job)).ID = 1 }).
4098-
Return(nil)
4099-
svc.orm.On("ApproveSpec",
4100-
mock.Anything,
4101-
spec.ID,
4102-
externalJobID,
4103-
).Return(nil)
4104-
svc.fmsClient.On("ApprovedJob",
4105-
mock.MatchedBy(func(ctx context.Context) bool { return true }),
4106-
&proto.ApprovedJobRequest{
4107-
Uuid: jp.RemoteUUID.String(),
4108-
Version: int64(spec.Version),
4109-
},
4110-
).Return(&proto.ApprovedJobResponse{}, nil)
4111-
svc.orm.On("CountJobProposalsByStatus", mock.Anything).Return(&feeds.JobProposalCounts{}, nil)
4112-
svc.orm.On("WithDataSource", mock.Anything).Return(feeds.ORM(svc.orm))
4113-
svc.jobORM.On("WithDataSource", mock.Anything).Return(job.ORM(svc.jobORM))
4028+
svc.orm.On("GetSpec", mock.Anything, cancelledSpec.ID).Return(cancelledSpec, nil)
4029+
svc.orm.On("GetJobProposal", mock.Anything, jp.ID).Return(jp, nil)
4030+
svc.orm.On("ListSpecsByJobProposalIDs", mock.Anything, []int64{cancelledSpec.JobProposalID}).
4031+
Return([]feeds.JobProposalSpec{otherSpec, *cancelledSpec}, nil)
41144032
},
4115-
id: spec.ID,
4116-
force: true,
4033+
id: cancelledSpec.ID,
4034+
force: false,
4035+
wantErr: "the job spec with version 2 is already approved",
41174036
},
41184037
{
4119-
name: "already existing FMS managed job replacement success if forced",
4120-
httpTimeout: commonconfig.MustNewDuration(1 * time.Minute),
4038+
name: "rejected spec failed cannot be approved",
41214039
before: func(svc *TestService) {
4122-
svc.connMgr.On("GetClient", jp.FeedsManagerID).Return(svc.fmsClient, nil)
4123-
svc.orm.On("GetSpec", mock.Anything, spec.ID).Return(spec, nil)
4040+
svc.orm.On("GetSpec", mock.Anything, cancelledSpec.ID).Return(rejectedSpec, nil)
41244041
svc.orm.On("GetJobProposal", mock.Anything, jp.ID).Return(jp, nil)
4125-
svc.jobORM.On("AssertBridgesExist", mock.Anything, mock.IsType(pipeline.Pipeline{})).Return(nil)
4126-
svc.orm.EXPECT().GetApprovedSpec(mock.Anything, jp.ID).Return(&feeds.JobProposalSpec{ID: 100}, nil)
4127-
svc.orm.EXPECT().CancelSpec(mock.Anything, int64(100)).Return(nil)
4128-
svc.jobORM.On("FindJobByExternalJobID", mock.Anything, externalJobID).Return(job.Job{}, sql.ErrNoRows)
4129-
svc.jobORM.On("FindJobIDByStreamID", mock.Anything, mock.Anything).Return(j.ID, nil)
4130-
svc.spawner.On("DeleteJob", mock.Anything, mock.Anything, j.ID).Return(nil)
4131-
4132-
svc.spawner.
4133-
On("CreateJob",
4134-
mock.Anything,
4135-
mock.Anything,
4136-
mock.MatchedBy(func(j *job.Job) bool {
4137-
return j.Name.String == streamName
4138-
}),
4139-
).
4140-
Run(func(args mock.Arguments) { (args.Get(2).(*job.Job)).ID = 1 }).
4141-
Return(nil)
4142-
svc.orm.On("ApproveSpec",
4143-
mock.Anything,
4144-
spec.ID,
4145-
externalJobID,
4146-
).Return(nil)
4147-
svc.fmsClient.On("ApprovedJob",
4148-
mock.MatchedBy(func(ctx context.Context) bool { return true }),
4149-
&proto.ApprovedJobRequest{
4150-
Uuid: jp.RemoteUUID.String(),
4151-
Version: int64(spec.Version),
4152-
},
4153-
).Return(&proto.ApprovedJobResponse{}, nil)
4154-
svc.orm.On("CountJobProposalsByStatus", mock.Anything).Return(&feeds.JobProposalCounts{}, nil)
4155-
svc.orm.On("WithDataSource", mock.Anything).Return(feeds.ORM(svc.orm))
4156-
svc.jobORM.On("WithDataSource", mock.Anything).Return(job.ORM(svc.jobORM))
41574042
},
4158-
id: spec.ID,
4159-
force: true,
4043+
id: rejectedSpec.ID,
4044+
force: false,
4045+
wantErr: "cannot approve a rejected spec",
41604046
},
41614047
{
41624048
name: "spec does not exist",
@@ -4190,7 +4076,7 @@ func Test_Service_ApproveSpec_Stream(t *testing.T) {
41904076
JobProposalID: jp.ID,
41914077
Status: feeds.SpecStatusRejected,
41924078
}
4193-
svc.orm.On("GetSpec", mock.Anything, rspec.ID, mock.Anything).Return(rspec, nil)
4079+
svc.orm.On("GetSpec", mock.Anything, rspec.ID).Return(rspec, nil)
41944080
svc.orm.On("GetJobProposal", mock.Anything, jp.ID).Return(jp, nil)
41954081
},
41964082
id: spec.ID,
@@ -4239,7 +4125,6 @@ func Test_Service_ApproveSpec_Stream(t *testing.T) {
42394125
svc.jobORM.On("AssertBridgesExist", mock.Anything, mock.IsType(pipeline.Pipeline{})).Return(nil)
42404126

42414127
svc.jobORM.On("FindJobByExternalJobID", mock.Anything, externalJobID).Return(job.Job{}, sql.ErrNoRows)
4242-
svc.jobORM.On("FindJobIDByStreamID", mock.Anything, mock.Anything).Return(int32(0), sql.ErrNoRows)
42434128

42444129
svc.spawner.
42454130
On("CreateJob",
@@ -4267,7 +4152,6 @@ func Test_Service_ApproveSpec_Stream(t *testing.T) {
42674152
svc.jobORM.On("AssertBridgesExist", mock.Anything, mock.IsType(pipeline.Pipeline{})).Return(nil)
42684153

42694154
svc.jobORM.On("FindJobByExternalJobID", mock.Anything, externalJobID).Return(job.Job{}, sql.ErrNoRows)
4270-
svc.jobORM.On("FindJobIDByStreamID", mock.Anything, mock.Anything).Return(int32(0), sql.ErrNoRows)
42714155

42724156
svc.spawner.
42734157
On("CreateJob",
@@ -4301,7 +4185,6 @@ func Test_Service_ApproveSpec_Stream(t *testing.T) {
43014185
svc.jobORM.On("AssertBridgesExist", mock.Anything, mock.IsType(pipeline.Pipeline{})).Return(nil)
43024186

43034187
svc.jobORM.On("FindJobByExternalJobID", mock.Anything, externalJobID).Return(job.Job{}, sql.ErrNoRows)
4304-
svc.jobORM.On("FindJobIDByStreamID", mock.Anything, mock.Anything).Return(int32(0), sql.ErrNoRows)
43054188

43064189
svc.spawner.
43074190
On("CreateJob",
@@ -5065,7 +4948,7 @@ func Test_Service_UpdateSpecDefinition(t *testing.T) {
50654948
name: "success",
50664949
before: func(svc *TestService) {
50674950
svc.orm.
5068-
On("GetSpec", mock.Anything, specID, mock.Anything).
4951+
On("GetSpec", mock.Anything, specID).
50694952
Return(spec, nil)
50704953
svc.orm.On("UpdateSpecDefinition", mock.Anything,
50714954
specID,
@@ -5079,7 +4962,7 @@ func Test_Service_UpdateSpecDefinition(t *testing.T) {
50794962
name: "does not exist",
50804963
before: func(svc *TestService) {
50814964
svc.orm.
5082-
On("GetSpec", mock.Anything, specID, mock.Anything).
4965+
On("GetSpec", mock.Anything, specID).
50834966
Return(nil, sql.ErrNoRows)
50844967
},
50854968
specID: specID,
@@ -5089,7 +4972,7 @@ func Test_Service_UpdateSpecDefinition(t *testing.T) {
50894972
name: "other get errors",
50904973
before: func(svc *TestService) {
50914974
svc.orm.
5092-
On("GetSpec", mock.Anything, specID, mock.Anything).
4975+
On("GetSpec", mock.Anything, specID).
50934976
Return(nil, errors.New("other db error"))
50944977
},
50954978
specID: specID,
@@ -5104,7 +4987,7 @@ func Test_Service_UpdateSpecDefinition(t *testing.T) {
51044987
}
51054988

51064989
svc.orm.
5107-
On("GetSpec", mock.Anything, specID, mock.Anything).
4990+
On("GetSpec", mock.Anything, specID).
51084991
Return(spec, nil)
51094992
},
51104993
specID: specID,

0 commit comments

Comments
 (0)