diff --git a/go/parts/bus.go b/go/parts/bus.go index 13f67cf..6dc9d4d 100644 --- a/go/parts/bus.go +++ b/go/parts/bus.go @@ -101,9 +101,11 @@ func BusArm(state string) func(id string, ttl time.Duration) error { // FollowIdle follows the holder's idle state (the one BusArm writes) once the bundle serves, and hands the end of each // key (the bus's marker, OpExpire) to expired, a Receiver's Expired: the event drops an idle transfer, no timer runs. -// said is told each outcome (dropped, or the error that stopped it) for the holder's log. It blocks; run it in its own -// goroutine. -func FollowIdle(state string, expired func(id string) (bool, error), said func(id string, dropped bool, err error)) { +// Once the watch is taken it hands every transfer open (open, such as IDs of the holder's Transfers) to expired too, so +// a key that ended while the bundle was down, whose marker no watch saw, still drops its transfer, and a key lost is +// armed again. said is told each outcome (dropped, or the error that stopped it) for the holder's log. It returns +// once the start's look is done; run it in its own goroutine. +func FollowIdle(state string, open func() ([]string, error), expired func(id string) (bool, error), said func(id string, dropped bool, err error)) { <-stdio.Served() stdio.State(state).Follow("", func(ch stdio.StateChange) error { if ch.Op != stdio.OpExpire { @@ -115,4 +117,38 @@ func FollowIdle(state string, expired func(id string) (bool, error), said func(i } return err }) + expireOpen(state, open, expired, said) +} + +// expireOpen is FollowIdle's look at its start: every open transfer handed to expired, as its key's end would. +func expireOpen(state string, open func() ([]string, error), expired func(id string) (bool, error), said func(id string, dropped bool, err error)) { + say := func(id string, dropped bool, err error) { + if said != nil && (dropped || err != nil) { + said(id, dropped, err) + } + } + ids, err := open() + if err != nil { + say("", false, fmt.Errorf("the open transfers cannot be read at the start of %s's watch, so one whose idle key ended while the bundle was down waits for a look by hand: %w", state, err)) + return + } + for _, id := range ids { + dropped, err := expired(id) + say(id, dropped, err) + } +} + +// IDs answers the ids of the transfers open in t: FollowIdle's open. +func IDs[M any](t Transfers[M]) func() ([]string, error) { + return func() ([]string, error) { + all, err := t.List() + if err != nil { + return nil, err + } + out := make([]string, 0, len(all)) + for _, x := range all { + out = append(out, x.ID) + } + return out, nil + } } diff --git a/go/parts/receive_test.go b/go/parts/receive_test.go index 760c35a..d4ee5df 100644 --- a/go/parts/receive_test.go +++ b/go/parts/receive_test.go @@ -273,5 +273,33 @@ func aHolderTakesAFileInParts(t *testing.T) { if pkg.Bound = func() (uint64, error) { return 0, errors.New("no binding") }; pkg.CapNow() != pkg.Cap { t.Errorf("CapNow with no binding read = %d, want Cap %d", pkg.CapNow(), pkg.Cap) } + pkg.Bound = nil + // At the start of the idle watch every open transfer is looked at (expireOpen): one whose key ended while the + // bundle was down is dropped, one still within its bound is armed again for the rest. + pkg.Arm = func(id string, ttl time.Duration) error { armed[id] = ttl; return nil } + for id := range pkg.Transfers.(memTransfers[packageShape]) { + _, _ = pkg.Drop(id) + } + _, _ = pkg.Open("t9", "builder", []Want{part}, packageShape{"m", "1"}) + now = now.Add(IdleAfter + time.Hour) + _, _ = pkg.Open("t10", "builder", []Want{part}, packageShape{"m", "1"}) + now = now.Add(time.Hour) + var dropped []string + expireOpen("idle", IDs(pkg.Transfers), pkg.Expired, func(id string, d bool, err error) { + if err != nil { + t.Errorf("start's look at %s: %v", id, err) + } + if d { + dropped = append(dropped, id) + } + }) + if !reflect.DeepEqual(dropped, []string{"t9"}) || armed["t10"] != IdleAfter-time.Hour { + t.Errorf("at the start: dropped %v, t10 armed for %s; want t9 dropped and t10 armed for %s", dropped, armed["t10"], IdleAfter-time.Hour) + } + var said error + expireOpen("idle", func() ([]string, error) { return nil, errors.New("bus down") }, pkg.Expired, func(_ string, _ bool, err error) { said = err }) + if said == nil || !strings.Contains(said.Error(), "bus down") { + t.Errorf("start's look with no list said %v, want the failure", said) + } }) }