MithrilJS Stream.combine() violates atomic update semantics.

[dvdkhlng]

Mithril version: "next" branch (rev dca44b1)
Browser and OS: Ubuntu / Firefox 80

The following unit-test code (using tinytest) illustrates the issue:

        var a = m.stream();
        var b = a.map(x => x+1);
        var c = b.map(x => x+1);

        var numUpdates = 0;
        
        var d = m.stream.lift((x, y) => { numUpdates++; return x*y }, b, c);
        eq(0, numUpdates);
        
        /* initial change already used to trigger two updates: after c() is
           updated, c() trigers d() and b() is already ready.

           then recursion exits, falling back into b's dependentStreams update
           loop, where b() triggers d() again. */
        a(5);
        eq(1, numUpdates);
        eq(6*7, d());

        /* second change should also only trigger a single update of d() */
        numUpdates = 0;
        a(6);
        eq(7*8, d());
        eq(1, numUpdates);

The last two eq() assertions WRT numUpdates are violated (numUpdates will be reported as 2). This directly contradicts the manual which states that "Computed properties in Mithril are updated atomically: streams that depend on multiple streams will never be called more than once per value update[..]"

This issue can be fixed by making sure that combine() only pushes the change downstreams, after all of the mappers received the update that was advertised by the preceding invokation of _changing().

The following patch solves the issue for me (but may not be good enough for the mithril codebase, my js coding skills are limited).

diff --git a/stream/stream.js b/stream/stream.js
index 98f41d96..308407af 100644
--- a/stream/stream.js
+++ b/stream/stream.js
@@ -110,15 +110,19 @@ function combine(fn, streams) {
 	var changed = []
 
 	var mappers = streams.map(function(s) {
-		return s._map(function(value) {
+		var mapper = s._map(function(value) {
 			changed.push(s)
-			if (ready || streams.every(function(s) { return s._state !== "pending" })) {
-				ready = true
+			var finalChange = mappers.every(function(s) { return mapper == s || s._state !== "changing" });
+			var gotReady = ready || streams.every(function(s) { return s._state === "active"; });
+			
+			if (finalChange && gotReady) {
 				stream(fn.apply(null, streams.concat([changed])))
 				changed = []
+				ready = true;
 			}
 			return value
 		}, true)
+                return mapper;
 	})
 
 	var endStream = stream.end.map(function(value) {

This is only one half of the problem. Stream.combine() does not propagate the _changing() signal down-streams. This leads to more atomic semantics mis-behaviour when chaining Stream.combine(). See this unit test that still fails in spite of the fix above:

        var a = m.stream();
        var b = a.map(x => x+1);
        var c = b.map(x => x+1);
        var d = m.stream.lift((x, y) => x*y, b, c);

        /* stream.combine() used to not forward the _changing() signal, thereby
         * breaking atomic-behaviour downstreams.  We provoke this here by
         * chaining the result of one combine() with another combine(). */
        var numUpdates = 0;
        var e = m.stream.lift((x, y) => { numUpdates++; return x+y }, c, d);

        a(5);
        eq(6*7 + 7, e())
        eq(1, numUpdates);

        numUpdates = 0;
        a(6);
        eq(7*8 + 8, e())
        eq(1, numUpdates);        

I can mitigate this problem by making stream.combine() forward the _changing() notification as follows:

diff --git a/stream/stream.js b/stream/stream.js
index 308407af..b2861111 100644
--- a/stream/stream.js
+++ b/stream/stream.js
@@ -120,6 +120,12 @@ function combine(fn, streams) {
 			}
 			return value
 		}, true)
+                /* override _changing to forward state change to downstream of
+                 * combine() */
+                mapper._changing = function() {
+		        if (open(mapper)) mapper._state = "changing"
+                        stream._changing();
+                }
                 return mapper;
 	})

Update: modified the first patch as it failed to pass the trivial test

        var a = m.stream(0);
        var b = m.stream(1);
        var c = m.stream.lift((aa, bb) => [a, b], a, b);  //m.stream.merge([a, b]);
        var numUpdates = m.stream.scan((acc,dummy) => acc+1, 0, c);
        a(5);
        eq(2, numUpdates());

[dvdkhlng]

Note that I never managed to fully fix Mithril's stream implementation WRT the above bugs, when considering all the interactions between streams returning SKIP and pending streams.

I think the crucial issue I failed to fix was that SKIP would create an inconsistent state: some nodes would be marked as "changing" fist, but due to SKIP would never get the final value ("finalChange").

If finally gave up and wrote a new stream implementation from scratch that represents stream dependencies as a graph and propagates changes breadth-first. My implementation is not exactly API compatible with Mithril streams, and I may not be able to share it here. It would also likely break existing code, as it is difficult to exactly reproduce the current implementation's semantics.