sampleCombine bug-like behavior

[nukisman]

Hey folks!
I found a strange behavior of sampleCombine.
It reproduces in case of: input.compose(sampleCombine(derivedFromInput, independentOfInput)) - only when:

• some derivedFromInput (or just same input) is the first parameter of sampleCombine
• and other parameters are independent of input.
  In such cases other parameters are independent of input will be one step late.

Code to reproduce:

xs.periodic(100)
  .take(5)
  .compose(all => { // all - independent of values
    // squaresAll - independent of values
    const squaresAll = all.map(v => v * v).debug('squareAll');

    // values - 'derived' from it self
    const values = all.debug('values'); 

    // squaresValues - derived from values
    const squaresValues = values.map(v => v * v).debug('squareValues'); 

    // [ 0, 0, 0 ] - OK
    // [ 2, 4, 2 ] - OK
    // [ 3, 9, 3 ] - OK
    // [ 4, 16, 4 ] - OK
    // return values.compose(sampleCombine(squaresAll, values));

    // [ 0, 0, 0 ] - OK
    // [ 2, 4, 2 ] - OK
    // [ 3, 9, 3 ] - OK
    // [ 4, 16, 4 ] - OK
    // return values.compose(sampleCombine(squaresValues, values));

    // [ 0, 0, 0 ] - OK
    // [ 2, 2, 4 ] - OK
    // [ 3, 3, 9 ] - OK
    // [ 4, 4, 16 ] - OK
    // return values.compose(sampleCombine(values, squaresValues));

    // [ 0, 0, 0 ] - missed,
    // [ 2, 2, 1 ] - incorrect (squaresAll: previous value)
    // [ 3, 3, 4 ] - incorrect (squaresAll: previous value)
    // [ 4, 4, 9 ] - incorrect (squaresAll: previous value)
    // return values.compose(sampleCombine(values, squaresAll));

    // [ 0, 0, 0 ] - OK
    // [ 1, 1, 1 ] - OK
    // [ 2, 2, 4 ] - OK
    // [ 3, 3, 9 ] - OK
    // [ 4, 4, 16 ] - OK
    // return values.compose(sampleCombine(all, squaresAll));

    // [ 0, 0, 0 ] - OK
    // [ 1, 1, 1 ] - OK
    // [ 2, 2, 4 ] - OK
    // [ 3, 3, 9 ] - OK
    // [ 4, 4, 16 ] - OK
    // return values.compose(sampleCombine(all, squaresValues));

    // [ 0, 0, 0 ] - OK
    // [ 1, 1, 1 ] - OK
    // [ 2, 4, 2 ] - OK
    // [ 3, 9, 3 ] - OK
    // [ 4, 16, 4 ] - OK
    // return values.compose(sampleCombine(squaresAll, all));

    // [ 0, 0, 0 ] - missed,
    // [ 1, 1, 0 ] - incorrect (all: previous value)
    // [ 2, 4, 1 ] - incorrect (all: previous value)
    // [ 3, 9, 2 ] - incorrect (all: previous value)
    // [ 4, 16, 3 ] - incorrect (all: previous value)
    // return values.compose(sampleCombine(squaresValues, all));

    // [ 0, 0, 0 ] - OK
    // [ 1, 1, 1 ] - OK
    // [ 2, 4, 4 ] - OK
    // [ 3, 9, 9 ] - OK
    // [ 4, 16, 16 ] - OK
    // return values.compose(sampleCombine(squaresAll, squaresValues));

    // [ 0, 0, 0 ] - missed,
    // [ 1, 1, 0 ] - incorrect (squaresAll: previous value)
    // [ 2, 4, 1 ] - incorrect (squaresAll: previous value)
    // [ 3, 9, 4 ] - incorrect (squaresAll: previous value)
    // [ 4, 16, 9 ] - incorrect (squaresAll: previous value)
    return values.compose(sampleCombine(squaresValues, squaresAll));
  })
  .addListener(printer());

export const printer = <T = any>(): Listener<T> => ({
  next: console.log,
  error: console.error,
  complete: () => console.log('completed'),
});

[staltz]

Hi! Thanks for the detailed report.

At a first glance, this looks a lot like a diamond situation. I suggest taking a look at issues #283 and #309 for examples of other diamond cases. It's outside of the scope of xstream (as is of RxJS's scope too) to do glitch-free functional reactive programming. In the general case it's not an easy problem to solve while maintaining good performance in JS.

[PEZO19]

Wondering about a minimal extension to xstream for glitch problem, partially. It wants to solve it not automatically, but with explicit "syncing" utilizing the fact that xstream has synchronous update semantics.

"Top level" / "tickFactory" (maybe factory?, but not only, because async operators like delay) operators, which lead to starting a sync tick could check if syncPostponed array is empty when the control flow gets back to them after updating the stream pipeline, and this syncPostponed array could be populated with a special operator syncPostpone (or just sync):

o$ = a$.combine(b$).compose(syncPostpone)

syncPostpone could just "eat everything" from the regular stream flow (not sending any next/error/complete message) until the control flow gets back to the tick starter node, which then calls these syncPostponed operators back. The last message (next, error, complete) syncPostpone operator has received might be the one it can resolve to.

The only thing to change in these special "tickFactory" operators is to call to check if this syncPostponed array is empty - after sending any message (next/error/complete).
And beyond new syncPostponed operator semantics (which seems to be really simple), just adding some recursive logic if syncPostponed array is not empty (so the stream pipeline can have multiple of these on the same subtree), but this part is new code, not changing existing special "tickFactory" operators.

Is there a similar solution to this? Or is it flawed? What am I missing? Seems reasonably easy to me.

@staltz Wdyt?