feat(CRE-329): high fidelity offchain job service client (#16527)

* feat(deployment): high fidelty offchain client

* fix tests

* cleanup

core/services/feeds/service.go

@@ -705,7 +705,7 @@ func (s *service) ProposeJob(ctx context.Context, args *ProposeJobArgs) (int64,
 
 		if exists {
 			// note: CLO auto-increments the version number on re-proposal, so this should never happen
-			return 0, errors.New("proposed job spec version already exists")
+			return 0, fmt.Errorf("external job id %s: version conflict: version %d already exists at job proposal id %d %v", args.RemoteUUID, args.Version, existing.ID, *existing)
 		}
 	}
 

core/services/feeds/service_test.go

@@ -1285,7 +1285,7 @@ func Test_Service_ProposeJob(t *testing.T) {
 				svc.orm.On("ExistsSpecByJobProposalIDAndVersion", mock.Anything, jpFluxMonitor.ID, argsFluxMonitor.Version).Return(true, nil)
 			},
 			args:    argsFluxMonitor,
-			wantErr: "proposed job spec version already exists",
+			wantErr: "version conflict",
 		},
 		{
 			name: "upsert error",

deployment/ccip/changeset/v1_5/cs_jobspec.go

@@ -90,7 +90,7 @@ func JobSpecsForLanesChangeset(env deployment.Environment, c JobSpecsForLanesCon
 				// JOBID will be empty if the proposal failed.
 				return deployment.ChangesetOutput{
 					Jobs: Jobs,
-				}, fmt.Errorf("failed to propose job: %w", err)
+				}, fmt.Errorf("failed to propose job %s: %w", job, err)
 			}
 			Jobs[len(Jobs)-1].JobID = res.Proposal.JobId
 		}

deployment/data-streams/changeset/jd_register_nodes_test.go

@@ -6,6 +6,8 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap/zapcore"
 
+	"github.com/smartcontractkit/chainlink-integrations/evm/testutils"
+	nodev1 "github.com/smartcontractkit/chainlink-protos/job-distributor/v1/node"
 	"github.com/smartcontractkit/chainlink/deployment"
 	"github.com/smartcontractkit/chainlink/deployment/common/changeset"
 	"github.com/smartcontractkit/chainlink/deployment/environment/memory"
@@ -14,19 +16,21 @@ import (
 
 func TestRegisterNodesWithJD(t *testing.T) {
 	t.Parallel()
+	ctx := testutils.Context(t)
 	lggr := logger.TestLogger(t)
 	e := memory.NewMemoryEnvironment(t, lggr, zapcore.InfoLevel, memory.MemoryEnvironmentConfig{Chains: 1, Nodes: 1})
 
-	nodeP2pKey := e.NodeIDs[0]
-
 	jobClient, ok := e.Offchain.(*memory.JobClient)
 	require.True(t, ok, "expected Offchain to be of type *memory.JobClient")
-	require.Lenf(t, jobClient.Nodes, 1, "expected exactly 1 node")
+
+	resp, err := jobClient.ListNodes(ctx, &nodev1.ListNodesRequest{})
+	require.NoError(t, err)
+	require.Lenf(t, resp.Nodes, 1, "expected exactly 1 node")
 	require.Emptyf(t, jobClient.RegisteredNodes, "no registered nodes expected")
 
-	csaKey := jobClient.Nodes[nodeP2pKey].Keys.CSA.PublicKeyString()
+	csaKey := resp.Nodes[0].GetPublicKey()
 
-	e, err := changeset.Apply(t, e, nil,
+	e, err = changeset.Apply(t, e, nil,
 		changeset.Configure(
 			deployment.CreateLegacyChangeSet(RegisterNodesWithJD),
 			RegisterNodesInput{