Skip to content

fix: apply invocation config to workflow Data after callback #606

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
danskmt wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix: apply invocation config to workflow Data after callback #606

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 32 additions & 2 deletions pkg/workflow/dataimpl.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,10 +66,18 @@ func withPayload(payload *interface{}) Option {
}
}

// WithConfiguration copies IN_MEMORY_THRESHOLD_BYTES and TEMP_DIR_PATH from
// config when each key resolves to a value (including AddDefaultValue). If
// TEMP_DIR_PATH is absent, d.tempDirPath is left unchanged (typically ""), and
// os.CreateTemp uses the process default temp directory.
func WithConfiguration(config configuration.Configuration) Option {

@robertolopezlopez robertolopezlopez Jun 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

in an ideal world (at least the one in my head) there should be a receiver func (d *DataImpl) WithConfiguration(...). As this may be problematic to implement now (too many usages off the task scope), I suggest the following:

func WithConfiguration(config configuration.Configuration) Option {
   return func(d *DataImpl) { d.configureFrom(config) }
}

func (d *DataImpl) configureFrom(config configuration.Configuration) {
   // copy in memory threshold and temp dir path if present in config
}

func (d *DataImpl) applyConfiguration(config configuration.Configuration) {
   ... your nil check
   d.configureFrom(config)
   ... rest of logic from your method
}

what do you think - this doesn't change the existing signatures which is probably used from other places, while keeping the state in the receiver

return func(d *DataImpl) {
d.inMemoryThreshold = config.GetInt(configuration.IN_MEMORY_THRESHOLD_BYTES)
d.tempDirPath = config.GetString(configuration.TEMP_DIR_PATH)
if v := config.Get(configuration.IN_MEMORY_THRESHOLD_BYTES); v != nil {

@basti-snyk basti-snyk Jul 15, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggestion: exported-API semantic change worth a direct-caller test. WithConfiguration previously set inMemoryThreshold/tempDirPath unconditionally (unset key → 0/""); it now skips a key when config.Get(...) == nil. This is a genuine fix (an unset threshold no longer means "spill everything"), but WithConfiguration is public and no test asserts the new behavior for direct external callers when one key is set and the other is unset. Any caller that relied on "unset threshold ⇒ 0 ⇒ spill" changes silently. Add a focused unit test for the mixed set/unset case. — AI review

d.inMemoryThreshold = config.GetInt(configuration.IN_MEMORY_THRESHOLD_BYTES)
}
if v := config.Get(configuration.TEMP_DIR_PATH); v != nil {
d.tempDirPath = config.GetString(configuration.TEMP_DIR_PATH)
}
}
}

Expand Down Expand Up @@ -256,6 +264,28 @@ func (d *DataImpl) AddError(err snyk_errors.Error) {
d.errors = append(d.errors, err)
}

// applyConfiguration re-evaluates the payload location using the given
// configuration. Field updates use the same rules as WithConfiguration.
// If the payload is currently in memory and exceeds the
// configured threshold, it is written to disk under the configured temp
// directory. This allows the engine to apply its configuration to Data
// objects that were created without WithConfiguration.
func (d *DataImpl) applyConfiguration(config configuration.Configuration) {
if config.Get(configuration.IN_MEMORY_THRESHOLD_BYTES) == nil {

@basti-snyk basti-snyk Jul 15, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should-Fix (correctness): new nil-config panic path. Invoke now unconditionally calls applyConfiguration(options.config) for every *DataImpl output, and this first line dereferences the config. When a caller passes a nil config — Invoke(id, WithConfig(nil)) or the deprecated InvokeWithConfig(id, nil) — options.config is nil, so this panics (nil interface method call). Before this PR that path was harmless: the engine only stored the config (newInvocationContext, SetConfiguration) and never dereferenced it. Root-cause fix is one line at the top of this method: if config == nil { return }. Please also add a test: Invoke with WithConfig(nil) and a workflow returning NewData(...) must not panic. — AI review

return
}

WithConfiguration(config)(d)

if d.payloadLocation.Type == InMemory && d.payload != nil {
d.payloadLocation = setPayloadLocation(d.identifier, d.inMemoryThreshold, d.tempDirPath, d.payload, d.logger)
if d.payloadLocation.Type == OnDisk {
d.logger.Debug().Msg("payload relocated to disk after applyConfiguration")
d.payload = nil
}
}
}

func setPayloadLocation(id Identifier, inMemoryThreshold int, tempDirPath string, payload interface{}, logger *zerolog.Logger) Location {
payloadLocation := Location{
Path: "",
Expand Down
135 changes: 135 additions & 0 deletions pkg/workflow/dataimpl_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import (
"github.com/snyk/error-catalog-golang-public/snyk_errors"
"github.com/snyk/go-application-framework/pkg/configuration"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)

func Test_NewDataFromInput(t *testing.T) {
Expand Down Expand Up @@ -194,6 +195,140 @@ func Test_NewData(t *testing.T) {
assert.Contains(t, actualFiles[0].Name(), expectedFileName)
})

t.Run("applyConfiguration relocates in-memory payload to disk", func(t *testing.T) {
tmpDir := t.TempDir()
logger := zerolog.Nop()

id := NewTypeIdentifier(NewWorkflowIdentifier("cmd"), "applytest")
payloadBytes := []byte("payload that should end up on disk after applyConfiguration")

data := NewData(id, "application/octet-stream", payloadBytes, WithLogger(&logger))

// Without WithConfiguration, payload stays in memory (threshold defaults to -1)
di, ok := data.(*DataImpl)
require.True(t, ok)
assert.Equal(t, InMemory, di.payloadLocation.Type)
assert.NotNil(t, di.payload)

// Now apply a configuration with threshold=0 — everything spills to disk
cfg := configuration.NewInMemory()
cfg.Set(configuration.IN_MEMORY_THRESHOLD_BYTES, 0)
cfg.Set(configuration.TEMP_DIR_PATH, tmpDir)
di.applyConfiguration(cfg)

assert.Equal(t, OnDisk, di.payloadLocation.Type)
assert.Nil(t, di.payload)
assert.Contains(t, di.payloadLocation.Path, tmpDir)

// GetPayload still returns the original bytes
result := data.GetPayload()
assert.Equal(t, payloadBytes, result)
})

t.Run("applyConfiguration is no-op when threshold is disabled", func(t *testing.T) {
logger := zerolog.Nop()
id := NewTypeIdentifier(NewWorkflowIdentifier("cmd"), "noop")
payloadBytes := []byte("stays in memory")

data := NewData(id, "text/plain", payloadBytes, WithLogger(&logger))
di, ok := data.(*DataImpl)
require.True(t, ok)

cfg := configuration.NewInMemory()
cfg.Set(configuration.IN_MEMORY_THRESHOLD_BYTES, -1)
di.applyConfiguration(cfg)

assert.Equal(t, InMemory, di.payloadLocation.Type)
assert.NotNil(t, di.payload)
})

t.Run("applyConfiguration is no-op when threshold key is not set in config", func(t *testing.T) {
logger := zerolog.Nop()
id := NewTypeIdentifier(NewWorkflowIdentifier("cmd"), "unset")
payloadBytes := []byte("should stay in memory because key is unset")

data := NewData(id, "text/plain", payloadBytes, WithLogger(&logger))
di, ok := data.(*DataImpl)
require.True(t, ok)
assert.Equal(t, -1, di.inMemoryThreshold)

cfg := configuration.NewInMemory()
require.Nil(t, cfg.Get(configuration.IN_MEMORY_THRESHOLD_BYTES))
di.applyConfiguration(cfg)

assert.Equal(t, -1, di.inMemoryThreshold)
assert.Equal(t, InMemory, di.payloadLocation.Type)
assert.NotNil(t, di.payload)
})

t.Run("applyConfiguration uses AddDefaultValue when key is not Set", func(t *testing.T) {
tmpDir := t.TempDir()
logger := zerolog.Nop()
id := NewTypeIdentifier(NewWorkflowIdentifier("cmd"), "adddefault")
payloadBytes := []byte("spill via default value functions")

data := NewData(id, "application/octet-stream", payloadBytes, WithLogger(&logger))
di, ok := data.(*DataImpl)
require.True(t, ok)

cfg := configuration.NewInMemory()
cfg.AddDefaultValue(configuration.IN_MEMORY_THRESHOLD_BYTES, configuration.StandardDefaultValueFunction(0))
cfg.AddDefaultValue(configuration.TEMP_DIR_PATH, configuration.StandardDefaultValueFunction(tmpDir))

assert.False(t, cfg.IsSet(configuration.IN_MEMORY_THRESHOLD_BYTES))
assert.False(t, cfg.IsSet(configuration.TEMP_DIR_PATH))
require.NotNil(t, cfg.Get(configuration.IN_MEMORY_THRESHOLD_BYTES))
require.NotNil(t, cfg.Get(configuration.TEMP_DIR_PATH))

di.applyConfiguration(cfg)

assert.Equal(t, OnDisk, di.payloadLocation.Type)
assert.Nil(t, di.payload)
assert.Contains(t, di.payloadLocation.Path, tmpDir)
})

t.Run("applyConfiguration with threshold only uses system temp when TEMP_DIR_PATH absent", func(t *testing.T) {
logger := zerolog.Nop()
id := NewTypeIdentifier(NewWorkflowIdentifier("cmd"), "notemp")
payloadBytes := []byte("spill with no temp path in config")

data := NewData(id, "application/octet-stream", payloadBytes, WithLogger(&logger))
di, ok := data.(*DataImpl)
require.True(t, ok)

cfg := configuration.NewInMemory()
cfg.Set(configuration.IN_MEMORY_THRESHOLD_BYTES, 0)
require.Nil(t, cfg.Get(configuration.TEMP_DIR_PATH))

di.applyConfiguration(cfg)

assert.Equal(t, OnDisk, di.payloadLocation.Type)
assert.Contains(t, di.payloadLocation.Path, os.TempDir())
})

t.Run("applyConfiguration is no-op when payload is already on disk", func(t *testing.T) {
tmpDir := t.TempDir()
logger := zerolog.Nop()

cfg := configuration.NewInMemory()
cfg.Set(configuration.IN_MEMORY_THRESHOLD_BYTES, 0)
cfg.Set(configuration.TEMP_DIR_PATH, tmpDir)

id := NewTypeIdentifier(NewWorkflowIdentifier("cmd"), "alreadyondisk")
payloadBytes := []byte("on disk from the start")

data := NewData(id, "application/octet-stream", payloadBytes, WithConfiguration(cfg), WithLogger(&logger))
di, ok := data.(*DataImpl)
require.True(t, ok)
assert.Equal(t, OnDisk, di.payloadLocation.Type)
originalPath := di.payloadLocation.Path

// applyConfiguration again should not relocate
di.applyConfiguration(cfg)
assert.Equal(t, OnDisk, di.payloadLocation.Type)
assert.Equal(t, originalPath, di.payloadLocation.Path)
})

t.Run("when configuration is not provided, filesystem cache is not used", func(t *testing.T) {
expectedConfig := configuration.NewInMemory()
expectedLogger := zerolog.Logger{}
Expand Down
39 changes: 39 additions & 0 deletions pkg/workflow/engine_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -598,3 +598,42 @@ func Test_EngineImpl_InvokeWithContext_DefaultContext(t *testing.T) {
assert.NoError(t, err)
assert.NotNil(t, receivedCtx)
}

func Test_Invoke_AppliesConfigurationToOutput(t *testing.T) {
tmpDir := t.TempDir()

config := configuration.New()
config.Set(configuration.IN_MEMORY_THRESHOLD_BYTES, 0)
config.Set(configuration.TEMP_DIR_PATH, tmpDir)

engine := NewWorkFlowEngine(config)

wfId := NewWorkflowIdentifier("threshold-test")
flagset := pflag.NewFlagSet("tt", pflag.ContinueOnError)

payload := []byte("this payload should be relocated to disk by the engine after Invoke")

_, err := engine.Register(wfId, ConfigurationOptionsFromFlagset(flagset), func(invocation InvocationContext, input []Data) ([]Data, error) {
// Create Data WITHOUT WithConfiguration — simulates what most callers do
id := NewTypeIdentifier(invocation.GetWorkflowIdentifier(), "testdata")
d := NewData(id, "application/octet-stream", payload)
return []Data{d}, nil
})
assert.NoError(t, err)
assert.NoError(t, engine.Init())

output, err := engine.Invoke(wfId)
assert.NoError(t, err)
assert.Len(t, output, 1)

// The engine should have applied its config, relocating the payload to disk
di, ok := output[0].(*DataImpl)
assert.True(t, ok)
assert.Equal(t, OnDisk, di.payloadLocation.Type)
assert.Contains(t, di.payloadLocation.Path, tmpDir)
assert.Nil(t, di.payload)

// GetPayload still returns the original bytes from disk
result := output[0].GetPayload()
assert.Equal(t, payload, result)
}
9 changes: 9 additions & 0 deletions pkg/workflow/engineimpl.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -338,6 +338,15 @@ func (e *EngineImpl) Invoke(
localLogger.Printf("Workflow Start")
output, err = callback(invocationCtx, options.input)
localLogger.Printf("Workflow End")

// Apply the engine's configuration to output data so that
// IN_MEMORY_THRESHOLD_BYTES and TEMP_DIR_PATH are respected
// even when workflows create Data without WithConfiguration.
for _, d := range output {

@basti-snyk basti-snyk Jul 15, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggestion: relocation runs even when the callback returned an error. This loop is outside any if err == nil guard, so when a workflow returns partial output alongside a non-nil err, that soon-to-be-discarded output is still spilled to disk, leaving orphan temp files for data nobody consumes. Guarding the loop with if err == nil avoids the wasted I/O. — AI review

if di, ok := d.(*DataImpl); ok {
di.applyConfiguration(options.config)

@basti-snyk basti-snyk Jul 15, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should-Fix (contract): engine silently overrides a workflow's explicit WithConfiguration. This loop applies the engine config to every returned *DataImpl, and applyConfiguration re-runs WithConfiguration, overwriting inMemoryThreshold/tempDirPath. A workflow that deliberately built its Data with NewData(..., WithConfiguration(customCfg)) (e.g. a higher threshold to keep a payload in memory on purpose) has that intent silently replaced by the engine defaults and force-spilled. applyConfiguration can't distinguish "never configured" (the case this PR targets) from "deliberately configured". Either apply only when the Data was never configured, or document on Invoke that engine config always wins last. This override path is also untested (new tests only cover Data created without WithConfiguration). — AI review

}
}
}
} else {
err = fmt.Errorf("workflow '%v' not found", id)
Expand Down