solver: fix pipe signaling on incoming updates

Signed-off-by: Tonis Tiigi <tonistiigi@gmail.com>

solver/edge.go

@@ -177,6 +177,9 @@ func (e *edge) finishIncoming(req pipe.Sender) {
 
 // updateIncoming updates the current value of incoming pipe request
 func (e *edge) updateIncoming(req pipe.Sender) {
+	if debugScheduler {
+		logrus.Debugf("updateIncoming %s %#v desired=%s", e.edge.Vertex.Name(), e.edgeState, req.Request().Payload.(*edgeRequest).desiredState)
+	}
 	req.Update(&e.edgeState)
 }
 

solver/internal/pipe/pipe.go

@@ -11,11 +11,15 @@ import (
 type channel struct {
 	OnSendCompletion func()
 	value            atomic.Value
-	lastValue        interface{}
+	lastValue        *wrappedValue
+}
+
+type wrappedValue struct {
+	value interface{}
 }
 
 func (c *channel) Send(v interface{}) {
-	c.value.Store(v)
+	c.value.Store(&wrappedValue{value: v})
 	if c.OnSendCompletion != nil {
 		c.OnSendCompletion()
 	}
@@ -23,11 +27,11 @@ func (c *channel) Send(v interface{}) {
 
 func (c *channel) Receive() (interface{}, bool) {
 	v := c.value.Load()
-	if c.lastValue == v {
+	if v == nil || v.(*wrappedValue) == c.lastValue {
 		return nil, false
 	}
-	c.lastValue = v
-	return v, true
+	c.lastValue = v.(*wrappedValue)
+	return v.(*wrappedValue).value, true
 }
 
 type Pipe struct {