Allow storing arbitrary data in a job

cluster/job_once.go

@@ -29,6 +29,10 @@ const (
 
 	// scheduleOnceJitter is the range of jitter to add to intervals to avoid contention issues
 	scheduleOnceJitter = 100 * time.Millisecond
+
+	// propsLimit is the maximum length in bytes of the json-representation of a job's props.
+	// It exists to prevent jon go rountines from consuming too much memory, as they are long running.
+	propsLimit = 10000
 )
 
 type JobOnceMetadata struct {
@@ -80,6 +84,15 @@ func newJobOnce(pluginAPI JobPluginAPI, key string, runAt time.Time, callback *s
 		return nil, errors.Wrap(err, "failed to create job mutex")
 	}
 
+	propsBytes, err := json.Marshal(props)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to marshal props")
+	}
+
+	if len(propsBytes) > propsLimit {
+		return nil, errors.Errorf("props length extends limit")
+	}
+
 	return &JobOnce{
 		pluginAPI:      pluginAPI,
 		clusterMutex:   mutex,

cluster/job_once_test.go

@@ -637,35 +637,49 @@ func TestScheduleOnceSequential(t *testing.T) {
 }
 
 func TestScheduleOnceProps(t *testing.T) {
-	s := GetJobOnceScheduler(newMockPluginAPI(t))
+	t.Run("confirm props are returned", func(t *testing.T) {
+		s := GetJobOnceScheduler(newMockPluginAPI(t))
+
+		jobKey := model.NewId()
+		jobProps := struct {
+			Foo string
+		}{
+			Foo: "some foo",
+		}
 
-	jobKey := model.NewId()
-	jobProps := struct {
-		Foo string
-	}{
-		Foo: "some foo",
-	}
+		var mut sync.Mutex
+		var called bool
+		callback := func(key string, props any) {
+			require.Equal(t, jobKey, key)
+			require.Equal(t, jobProps, props)
+			mut.Lock()
+			defer mut.Unlock()
+			called = true
+		}
 
-	var mut sync.Mutex
-	var called bool
-	callback := func(key string, props any) {
-		require.Equal(t, jobKey, key)
-		require.Equal(t, jobProps, props)
-		mut.Lock()
-		defer mut.Unlock()
-		called = true
-	}
+		err := s.SetCallback(callback)
+		require.NoError(t, err)
+		if !s.started {
+			err = s.Start()
+			require.NoError(t, err)
+		}
 
-	err := s.SetCallback(callback)
-	require.NoError(t, err)
-	if !s.started {
-		err = s.Start()
+		_, err = s.ScheduleOnce(jobKey, time.Now().Add(100*time.Millisecond), jobProps)
 		require.NoError(t, err)
-	}
 
-	_, err = s.ScheduleOnce(jobKey, time.Now().Add(100*time.Millisecond), jobProps)
-	require.NoError(t, err)
+		// Check if callback was called
+		require.Eventually(t, func() bool { mut.Lock(); defer mut.Unlock(); return called }, time.Second, 50*time.Millisecond)
+	})
+
+	t.Run("props to large", func(t *testing.T) {
+		s := GetJobOnceScheduler(newMockPluginAPI(t))
 
-	// Check if callback was called
-	require.Eventually(t, func() bool { mut.Lock(); defer mut.Unlock(); return called }, time.Second, 50*time.Millisecond)
+		props := make([]byte, propsLimit)
+		for i := 0; i < propsLimit; i++ {
+			props[i] = 'a'
+		}
+
+		_, err := s.ScheduleOnce(model.NewId(), time.Now().Add(100*time.Millisecond), props)
+		require.Error(t, err)
+	})
 }