From be2f7aeaf12833ca1f12fc4ca1d126df6d76419d Mon Sep 17 00:00:00 2001 From: Jozef Kralik Date: Tue, 7 Jul 2026 11:45:36 +0200 Subject: [PATCH] fix(blockwise): return 4.08 on incomplete Block1 and make receive cache limit configurable --- dtls/client.go | 4 +- dtls/server/server.go | 4 +- net/blockwise/blockwise.go | 405 ++++++++++++++++++++++---------- net/blockwise/blockwise_test.go | 87 +++++++ options/commonOptions.go | 31 +++ options/commonOptions_test.go | 10 + options/config/common.go | 28 ++- tcp/client.go | 4 +- tcp/server/server.go | 4 +- udp/blockwise_test.go | 142 +++++++++++ udp/client.go | 4 +- udp/server/server.go | 7 +- 12 files changed, 590 insertions(+), 140 deletions(-) diff --git a/dtls/client.go b/dtls/client.go index 55960267..ce70146f 100644 --- a/dtls/client.go +++ b/dtls/client.go @@ -97,7 +97,7 @@ func Client(conn *dtls.Conn, opts ...udp.Option) *udpClient.Conn { if cfg.BlockwiseEnable { createBlockWise = func(cc *udpClient.Conn) *blockwise.BlockWise[*udpClient.Conn] { v := cc - return blockwise.New( + bw := blockwise.New( v, cfg.BlockwiseTransferTimeout, cfg.Errors, @@ -105,6 +105,8 @@ func Client(conn *dtls.Conn, opts ...udp.Option) *udpClient.Conn { return v.GetObservationRequest(token) }, ) + bw.SetReceivingMessagesCacheMaxEntries(cfg.BlockwiseReceivingMessagesCacheMaxEntries) + return bw } } diff --git a/dtls/server/server.go b/dtls/server/server.go index 99abf517..11974e8a 100644 --- a/dtls/server/server.go +++ b/dtls/server/server.go @@ -218,7 +218,7 @@ func (s *Server) createConn(connection *coapNet.Conn, inactivityMonitor udpClien if s.cfg.BlockwiseEnable { createBlockWise = func(cc *udpClient.Conn) *blockwise.BlockWise[*udpClient.Conn] { v := cc - return blockwise.New( + bw := blockwise.New( v, s.cfg.BlockwiseTransferTimeout, s.cfg.Errors, @@ -226,6 +226,8 @@ func (s *Server) createConn(connection *coapNet.Conn, inactivityMonitor udpClien return v.GetObservationRequest(token) }, ) + bw.SetReceivingMessagesCacheMaxEntries(s.cfg.BlockwiseReceivingMessagesCacheMaxEntries) + return bw } } session := NewSession( diff --git a/net/blockwise/blockwise.go b/net/blockwise/blockwise.go index 33bc53b5..9591540b 100644 --- a/net/blockwise/blockwise.go +++ b/net/blockwise/blockwise.go @@ -44,6 +44,8 @@ const ( moreBlocksFollowingMask = 0x8 // szxMask last 3bits represents SZX (SZX) szxMask = 0x7 + // max number of concurrent in-flight incoming block-wise assemblies. + DefaultReceivingMessagesCacheMaxEntries = 1024 ) // SZX enum representation for the size of the block: https://tools.ietf.org/html/rfc7959#section-2.2 @@ -135,12 +137,13 @@ type Client interface { } type BlockWise[C Client] struct { - cc C - receivingMessagesCache *cache.Cache[uint64, *messageGuard] - sendingMessagesCache *cache.Cache[uint64, *pool.Message] - errors func(error) - getSentRequestFromOutside func(token message.Token) (*pool.Message, bool) - expiration time.Duration + cc C + receivingMessagesCache *cache.Cache[uint64, *messageGuard] + receivingMessagesCacheMaxEntries int + sendingMessagesCache *cache.Cache[uint64, *pool.Message] + errors func(error) + getSentRequestFromOutside func(token message.Token) (*pool.Message, bool) + expiration time.Duration } type messageGuard struct { @@ -148,6 +151,8 @@ type messageGuard struct { *semaphore.Weighted } +var errReceivingMessagesCacheFull = errors.New("receiving messages cache is full") + func newRequestGuard(request *pool.Message) *messageGuard { return &messageGuard{ Message: request, @@ -167,15 +172,25 @@ func New[C Client]( getSentRequestFromOutside = func(message.Token) (*pool.Message, bool) { return nil, false } } return &BlockWise[C]{ - cc: cc, - receivingMessagesCache: cache.NewCache[uint64, *messageGuard](), - sendingMessagesCache: cache.NewCache[uint64, *pool.Message](), - errors: errors, - getSentRequestFromOutside: getSentRequestFromOutside, - expiration: expiration, + cc: cc, + receivingMessagesCache: cache.NewCache[uint64, *messageGuard](), + receivingMessagesCacheMaxEntries: DefaultReceivingMessagesCacheMaxEntries, + sendingMessagesCache: cache.NewCache[uint64, *pool.Message](), + errors: errors, + getSentRequestFromOutside: getSentRequestFromOutside, + expiration: expiration, } } +// SetReceivingMessagesCacheMaxEntries sets max entries for in-flight incoming +// block-wise assemblies. Value 0 means unlimited. +func (b *BlockWise[C]) SetReceivingMessagesCacheMaxEntries(maxEntries int) { + if maxEntries < 0 { + maxEntries = 0 + } + b.receivingMessagesCacheMaxEntries = maxEntries +} + func bufferSize(szx SZX, maxMessageSize uint32) int64 { if szx < SZXBERT { return szx.Size() @@ -321,6 +336,14 @@ func (b *BlockWise[C]) sendEntityIncomplete(w *responsewriter.ResponseWriter[C], w.SetMessage(sendMessage) } +func (b *BlockWise[C]) sendServiceUnavailable(w *responsewriter.ResponseWriter[C], token message.Token) { + sendMessage := b.cc.AcquireMessage(w.Message().Context()) + sendMessage.SetCode(codes.ServiceUnavailable) + sendMessage.SetToken(token) + sendMessage.SetType(message.NonConfirmable) + w.SetMessage(sendMessage) +} + func wantsToBeReceived(r *pool.Message) bool { hasBlock1 := r.HasOption(message.Block1) hasBlock2 := r.HasOption(message.Block2) @@ -346,30 +369,51 @@ func (b *BlockWise[C]) getSendingMessageCode(token uint64) (codes.Code, bool) { return v.Data().Code(), true } +type receiveRequestCall[C Client] struct { + w *responsewriter.ResponseWriter[C] + r *pool.Message + next func(w *responsewriter.ResponseWriter[C], r *pool.Message) +} + +func (c receiveRequestCall[C]) forward() { + c.next(c.w, c.r) +} + +func (b *BlockWise[C]) handleReceivedMessageWithErrorResponse(call receiveRequestCall[C], maxSZX SZX, maxMessageSize uint32) { + err := b.handleReceivedMessage(call.w, call.r, maxSZX, maxMessageSize, call.next) + if err == nil { + return + } + token := call.r.Token() + if errors.Is(err, errReceivingMessagesCacheFull) { + b.sendServiceUnavailable(call.w, token) + } else { + b.sendEntityIncomplete(call.w, token) + } + b.errors(fmt.Errorf("handleReceivedMessage(%v): %w", call.r, err)) +} + // Handle middleware which constructs COAP request from blockwise transfer and send COAP response via blockwise. func (b *BlockWise[C]) Handle(w *responsewriter.ResponseWriter[C], r *pool.Message, maxSZX SZX, maxMessageSize uint32, next func(w *responsewriter.ResponseWriter[C], r *pool.Message)) { if maxSZX > SZXBERT { panic("invalid maxSZX") } + call := receiveRequestCall[C]{ + w: w, + r: r, + next: next, + } token := r.Token() if len(token) == 0 { - err := b.handleReceivedMessage(w, r, maxSZX, maxMessageSize, next) - if err != nil { - b.sendEntityIncomplete(w, token) - b.errors(fmt.Errorf("handleReceivedMessage(%v): %w", r, err)) - } + b.handleReceivedMessageWithErrorResponse(call, maxSZX, maxMessageSize) return } tokenStr := token.Hash() sendingMessageCode, sendingMessageExist := b.getSendingMessageCode(tokenStr) if !sendingMessageExist || wantsToBeReceived(r) { - err := b.handleReceivedMessage(w, r, maxSZX, maxMessageSize, next) - if err != nil { - b.sendEntityIncomplete(w, token) - b.errors(fmt.Errorf("handleReceivedMessage(%v): %w", r, err)) - } + b.handleReceivedMessageWithErrorResponse(call, maxSZX, maxMessageSize) return } more, err := b.continueSendingMessage(w, r, maxSZX, maxMessageSize, sendingMessageCode) @@ -680,28 +724,51 @@ func copyToPayloadFromOffset(r *pool.Message, payloadFile *memfile.File, offset return payloadSize, nil } -func (b *BlockWise[C]) getCachedReceivedMessage(mg *messageGuard, r *pool.Message, tokenStr uint64, validUntil time.Time) (*pool.Message, func(), error) { - cannotLockError := func(err error) error { +func acquireMessageGuard(mg *messageGuard) error { + if err := mg.Acquire(mg.Context(), 1); err != nil { return fmt.Errorf("processReceivedMessage: cannot lock message: %w", err) } - if mg != nil { - errA := mg.Acquire(mg.Context(), 1) - if errA != nil { - return nil, nil, cannotLockError(errA) - } - return mg.Message, func() { mg.Release(1) }, nil + return nil +} + +func (b *BlockWise[C]) ensureReceivingMessagesCacheCapacity() error { + if b.receivingMessagesCacheMaxEntries <= 0 { + return nil } + if b.receivingMessagesCache.Length() < b.receivingMessagesCacheMaxEntries { + return nil + } + b.receivingMessagesCache.CheckExpirations(time.Now()) + if b.receivingMessagesCache.Length() >= b.receivingMessagesCacheMaxEntries { + return errReceivingMessagesCacheFull + } + return nil +} + +func newCloseStack() (appendFn func(func()), closeFn func()) { closeFnList := []func(){} - appendToClose := func(m *messageGuard) { - closeFnList = append(closeFnList, func() { - m.Release(1) - }) + appendFn = func(fn func()) { + closeFnList = append(closeFnList, fn) } - closeFn := func() { + closeFn = func() { for i := range closeFnList { closeFnList[len(closeFnList)-1-i]() } } + return appendFn, closeFn +} + +func (b *BlockWise[C]) getCachedReceivedMessage(mg *messageGuard, r *pool.Message, tokenStr uint64, validUntil time.Time) (*pool.Message, func(), error) { + if mg != nil { + if err := acquireMessageGuard(mg); err != nil { + return nil, nil, err + } + return mg.Message, func() { mg.Release(1) }, nil + } + if err := b.ensureReceivingMessagesCacheCapacity(); err != nil { + return nil, nil, err + } + appendClose, closeFn := newCloseStack() msg := b.cc.AcquireMessage(r.Context()) msg.ResetOptionsTo(r.Options()) msg.SetToken(r.Token()) @@ -709,11 +776,10 @@ func (b *BlockWise[C]) getCachedReceivedMessage(mg *messageGuard, r *pool.Messag msg.SetBody(memfile.New(make([]byte, 0, 1024))) msg.SetCode(r.Code()) mg = newRequestGuard(msg) - errA := mg.Acquire(mg.Context(), 1) - if errA != nil { - return nil, nil, cannotLockError(errA) + if err := acquireMessageGuard(mg); err != nil { + return nil, nil, err } - appendToClose(mg) + appendClose(func() { mg.Release(1) }) element, loaded := b.receivingMessagesCache.LoadOrStore(tokenStr, cache.NewElement(mg, validUntil, func(d *messageGuard) { if d == nil { return @@ -727,125 +793,224 @@ func (b *BlockWise[C]) getCachedReceivedMessage(mg *messageGuard, r *pool.Messag closeFn() return nil, nil, errors.New("request was already stored in cache") } - errA := mg.Acquire(mg.Context(), 1) - if errA != nil { + if err := acquireMessageGuard(mg); err != nil { closeFn() - return nil, nil, cannotLockError(errA) + return nil, nil, err } - appendToClose(mg) + appendClose(func() { mg.Release(1) }) } return mg.Message, closeFn, nil } -//nolint:gocyclo,gocognit -func (b *BlockWise[C]) processReceivedMessage(w *responsewriter.ResponseWriter[C], r *pool.Message, maxSzx SZX, next func(w *responsewriter.ResponseWriter[C], r *pool.Message), blockType message.OptionID, sizeType message.OptionID) error { - token := r.Token() - if len(token) == 0 { +func getRequestBlock[C Client](r *pool.Message, blockType message.OptionID, next func(w *responsewriter.ResponseWriter[C], r *pool.Message), w *responsewriter.ResponseWriter[C]) (szx SZX, num int64, more bool, done bool, err error) { + if len(r.Token()) == 0 || r.Code() == codes.GET || r.Code() == codes.DELETE { next(w, r) - return nil - } - if r.Code() == codes.GET || r.Code() == codes.DELETE { - next(w, r) - return nil + return 0, 0, false, true, nil } block, err := r.GetOptionUint32(blockType) if err != nil { if errors.Is(err, message.ErrOptionNotFound) { next(w, r) - return nil + return 0, 0, false, true, nil } - return fmt.Errorf("cannot get Block(optionID=%d) option: %w", blockType, err) + return 0, 0, false, false, fmt.Errorf("cannot get Block(optionID=%d) option: %w", blockType, err) } - szx, num, more, err := DecodeBlockOption(block) + szx, num, more, err = DecodeBlockOption(block) if err != nil { - return fmt.Errorf("cannot decode block option: %w", err) - } - sentRequest := b.getSentRequest(token) - if sentRequest != nil { - defer b.cc.ReleaseMessage(sentRequest) + return 0, 0, false, false, fmt.Errorf("cannot decode block option: %w", err) } + return szx, num, more, false, nil +} + +func (b *BlockWise[C]) resolveReceivedToken(r *pool.Message, sentRequest *pool.Message, blockType message.OptionID) (message.Token, time.Time, error) { + token := r.Token() validUntil := b.getValidUntil(sentRequest) if blockType == message.Block2 && sentRequest == nil { - return errors.New("cannot request body without paired request") + return nil, time.Time{}, errors.New("cannot request body without paired request") } - if isObserveResponse(r) { - token, validUntil, err = b.handleObserveResponse(sentRequest) - if err != nil { - return fmt.Errorf("cannot process message: %w", err) - } + if !isObserveResponse(r) { + return token, validUntil, nil + } + observeToken, observeValidUntil, err := b.handleObserveResponse(sentRequest) + if err != nil { + return nil, time.Time{}, fmt.Errorf("cannot process message: %w", err) } + return observeToken, observeValidUntil, nil +} - tokenStr := token.Hash() - var cachedReceivedMessageGuard *messageGuard - if e := b.receivingMessagesCache.Load(tokenStr); e != nil { - cachedReceivedMessageGuard = e.Data() +func tryForwardWithoutCache[C Client](cachedGuard *messageGuard, more bool, num int64, szx SZX, maxSzx SZX, call receiveRequestCall[C]) (SZX, bool, error) { + szx = getSzx(szx, maxSzx) + if cachedGuard != nil || more { + return szx, false, nil } - if cachedReceivedMessageGuard == nil { - szx = getSzx(szx, maxSzx) - // if there is no more then just forward req to next handler - if !more { - next(w, r) - return nil + if num != 0 { + return szx, false, fmt.Errorf("received incomplete body: expected block number 0 for single-block transfer, got %v", num) + } + call.forward() + return szx, true, nil +} + +type receivedFinalizeState struct { + tokenStr uint64 + token message.Token + cachedReceivedMessage *pool.Message + blockType message.OptionID + sizeType message.OptionID + msgType message.Type +} + +func (b *BlockWise[C]) finalizeReceivedRequest(state receivedFinalizeState, call receiveRequestCall[C]) error { + b.receivingMessagesCache.Delete(state.tokenStr) + state.cachedReceivedMessage.Remove(state.blockType) + state.cachedReceivedMessage.Remove(state.sizeType) + state.cachedReceivedMessage.SetType(state.msgType) + if !bytes.Equal(state.cachedReceivedMessage.Token(), state.token) { + b.sendingMessagesCache.Delete(state.tokenStr) + } + if _, err := state.cachedReceivedMessage.Body().Seek(0, io.SeekStart); err != nil { + return fmt.Errorf("cannot seek to start of cachedReceivedMessage request: %w", err) + } + call.next(call.w, state.cachedReceivedMessage) + return nil +} + +type receivedContinueState struct { + token message.Token + blockType message.OptionID + sentRequest *pool.Message + num int64 + payloadSize int64 + szx SZX + more bool +} + +func (b *BlockWise[C]) continueReceivedRequest(call receiveRequestCall[C], state receivedContinueState) error { + sendMessage := b.cc.AcquireMessage(call.r.Context()) + sendMessage.SetToken(state.token) + if state.blockType == message.Block2 { + state.num = state.payloadSize / state.szx.Size() + sendMessage.ResetOptionsTo(state.sentRequest.Options()) + sendMessage.SetCode(state.sentRequest.Code()) + sendMessage.Remove(message.Observe) + sendMessage.Remove(message.Block1) + sendMessage.Remove(message.Size1) + respBlock, err := EncodeBlockOption(state.szx, state.num, state.more) + if err != nil { + b.cc.ReleaseMessage(sendMessage) + return fmt.Errorf("cannot encode block option(%v,%v,%v): %w", state.szx, state.num, state.more, err) } + sendMessage.SetOptionUint32(state.blockType, respBlock) + call.w.SetMessage(sendMessage) + return nil } - cachedReceivedMessage, closeCachedReceivedMessage, err := b.getCachedReceivedMessage(cachedReceivedMessageGuard, r, tokenStr, validUntil) + + sendMessage.SetCode(codes.Continue) + respBlock, err := EncodeBlockOption(state.szx, state.num, state.more) if err != nil { - return err + b.cc.ReleaseMessage(sendMessage) + return fmt.Errorf("cannot encode block option(%v,%v,%v): %w", state.szx, state.num, state.more, err) } - defer closeCachedReceivedMessage() + sendMessage.SetOptionUint32(state.blockType, respBlock) + call.w.SetMessage(sendMessage) + return nil +} - defer func(err *error) { - if *err != nil { - b.receivingMessagesCache.Delete(tokenStr) - } - }(&err) +func releaseMessageIfNotNil[C Client](cc C, msg *pool.Message) { + if msg != nil { + cc.ReleaseMessage(msg) + } +} + +func cachedReceivedMessageGuardByToken[C Client](b *BlockWise[C], tokenStr uint64) *messageGuard { + if e := b.receivingMessagesCache.Load(tokenStr); e != nil { + return e.Data() + } + return nil +} + +func (b *BlockWise[C]) copyReceivedPayload(r, cachedReceivedMessage *pool.Message) (*memfile.File, int64, error) { payloadFile, payloadSize, err := b.getPayloadFromCachedReceivedMessage(r, cachedReceivedMessage) if err != nil { - return fmt.Errorf("cannot get payload: %w", err) + return nil, 0, fmt.Errorf("cannot get payload: %w", err) } - off := num * szx.Size() - if off == payloadSize { //nolint:nestif - payloadSize, err = copyToPayloadFromOffset(r, payloadFile, off) + return payloadFile, payloadSize, nil +} + +func (b *BlockWise[C]) continueOrFinalizeReceivedRequest(call receiveRequestCall[C], state receivedContinueState, tokenStr uint64, sizeType message.OptionID, cachedReceivedMessage *pool.Message) error { + off := state.num * state.szx.Size() + payloadFile, payloadSize, err := b.copyReceivedPayload(call.r, cachedReceivedMessage) + if err != nil { + return err + } + if off == payloadSize { + payloadSize, err = copyToPayloadFromOffset(call.r, payloadFile, off) if err != nil { return fmt.Errorf("cannot copy data to payload: %w", err) } - if !more { - b.receivingMessagesCache.Delete(tokenStr) - cachedReceivedMessage.Remove(blockType) - cachedReceivedMessage.Remove(sizeType) - cachedReceivedMessage.SetType(r.Type()) - if !bytes.Equal(cachedReceivedMessage.Token(), token) { - b.sendingMessagesCache.Delete(tokenStr) - } - _, errS := cachedReceivedMessage.Body().Seek(0, io.SeekStart) - if errS != nil { - return fmt.Errorf("cannot seek to start of cachedReceivedMessage request: %w", errS) - } - next(w, cachedReceivedMessage) - return nil + if !state.more { + return b.finalizeReceivedRequest(receivedFinalizeState{ + tokenStr: tokenStr, + token: state.token, + cachedReceivedMessage: cachedReceivedMessage, + blockType: state.blockType, + sizeType: sizeType, + msgType: call.r.Type(), + }, call) } } + if off != payloadSize && !state.more { + return fmt.Errorf("received incomplete body: missing previous blocks (offset=%v, payloadSize=%v)", off, payloadSize) + } + state.payloadSize = payloadSize + return b.continueReceivedRequest(call, state) +} - szx = getSzx(szx, maxSzx) - sendMessage := b.cc.AcquireMessage(r.Context()) - sendMessage.SetToken(token) - if blockType == message.Block2 { - num = payloadSize / szx.Size() - sendMessage.ResetOptionsTo(sentRequest.Options()) - sendMessage.SetCode(sentRequest.Code()) - sendMessage.Remove(message.Observe) - sendMessage.Remove(message.Block1) - sendMessage.Remove(message.Size1) - } else { - sendMessage.SetCode(codes.Continue) +func (b *BlockWise[C]) processReceivedMessage(w *responsewriter.ResponseWriter[C], r *pool.Message, maxSzx SZX, next func(w *responsewriter.ResponseWriter[C], r *pool.Message), blockType message.OptionID, sizeType message.OptionID) (err error) { + call := receiveRequestCall[C]{ + w: w, + r: r, + next: next, + } + szx, num, more, done, err := getRequestBlock(r, blockType, next, w) + if done { + return nil } - respBlock, err := EncodeBlockOption(szx, num, more) if err != nil { - b.cc.ReleaseMessage(sendMessage) - return fmt.Errorf("cannot encode block option(%v,%v,%v): %w", szx, num, more, err) + return err } - sendMessage.SetOptionUint32(blockType, respBlock) - w.SetMessage(sendMessage) - return nil + token := r.Token() + sentRequest := b.getSentRequest(token) + defer releaseMessageIfNotNil(b.cc, sentRequest) + token, validUntil, err := b.resolveReceivedToken(r, sentRequest, blockType) + if err != nil { + return err + } + + tokenStr := token.Hash() + cachedReceivedMessageGuard := cachedReceivedMessageGuardByToken(b, tokenStr) + szx, done, err = tryForwardWithoutCache(cachedReceivedMessageGuard, more, num, szx, maxSzx, call) + if done || err != nil { + return err + } + cachedReceivedMessage, closeCachedReceivedMessage, err := b.getCachedReceivedMessage(cachedReceivedMessageGuard, r, tokenStr, validUntil) + if err != nil { + return err + } + defer closeCachedReceivedMessage() + + defer func() { + if err != nil { + b.receivingMessagesCache.Delete(tokenStr) + } + }() + continueState := receivedContinueState{ + token: token, + blockType: blockType, + sentRequest: sentRequest, + num: num, + szx: getSzx(szx, maxSzx), + more: more, + } + return b.continueOrFinalizeReceivedRequest(call, continueState, tokenStr, sizeType, cachedReceivedMessage) } diff --git a/net/blockwise/blockwise_test.go b/net/blockwise/blockwise_test.go index 70d173ff..a4e59ad2 100644 --- a/net/blockwise/blockwise_test.go +++ b/net/blockwise/blockwise_test.go @@ -615,3 +615,90 @@ func TestBlockWiseWriteTestMessage(t *testing.T) { }) } } + +func TestProcessReceivedMessageReturnsEntityIncompleteForMissingPreviousFinalBlock(t *testing.T) { + bw := New(newTestClient(), time.Second*3600, func(error) {}, nil) + ctx := context.Background() + token := message.Token{0xAA} + + block0, err := EncodeBlockOption(SZX16, 0, true) + require.NoError(t, err) + first := bw.cc.AcquireMessage(ctx) + first.SetCode(codes.POST) + first.SetToken(token) + first.SetOptionUint32(message.Block1, block0) + first.SetBody(bytes.NewReader(make([]byte, int(SZX16.Size())))) + w1 := responsewriter.New(bw.cc.AcquireMessage(ctx), bw.cc) + bw.Handle(w1, first, SZX16, uint32(SZX16.Size()), func(_ *responsewriter.ResponseWriter[*testClient], _ *pool.Message) { + require.Fail(t, "next must not be called while more blocks follow") + }) + require.Equal(t, codes.Continue, w1.Message().Code()) + + block2Final, err := EncodeBlockOption(SZX16, 2, false) + require.NoError(t, err) + second := bw.cc.AcquireMessage(ctx) + second.SetCode(codes.POST) + second.SetToken(token) + second.SetOptionUint32(message.Block1, block2Final) + second.SetBody(bytes.NewReader(make([]byte, int(SZX16.Size())))) + w2 := responsewriter.New(bw.cc.AcquireMessage(ctx), bw.cc) + bw.Handle(w2, second, SZX16, uint32(SZX16.Size()), func(_ *responsewriter.ResponseWriter[*testClient], _ *pool.Message) { + require.Fail(t, "next must not be called for incomplete transfer") + }) + require.Equal(t, codes.RequestEntityIncomplete, w2.Message().Code()) + require.Empty(t, bw.receivingMessagesCache.LoadAndDeleteAll()) +} + +func TestProcessReceivedMessageReturnsEntityIncompleteForSingleFinalNonZeroBlock(t *testing.T) { + bw := New(newTestClient(), time.Second*3600, func(error) {}, nil) + ctx := context.Background() + block2Final, err := EncodeBlockOption(SZX16, 2, false) + require.NoError(t, err) + + req := bw.cc.AcquireMessage(ctx) + req.SetCode(codes.POST) + req.SetToken(message.Token{0xAB}) + req.SetOptionUint32(message.Block1, block2Final) + req.SetBody(bytes.NewReader(make([]byte, int(SZX16.Size())))) + + w := responsewriter.New(bw.cc.AcquireMessage(ctx), bw.cc) + bw.Handle(w, req, SZX16, uint32(SZX16.Size()), func(_ *responsewriter.ResponseWriter[*testClient], _ *pool.Message) { + require.Fail(t, "next must not be called for incomplete transfer") + }) + require.Equal(t, codes.RequestEntityIncomplete, w.Message().Code()) +} + +func TestReceivingMessagesCacheIsBounded(t *testing.T) { + var gotErr error + bw := New(newTestClient(), time.Second*3600, func(err error) { gotErr = err }, nil) + bw.receivingMessagesCacheMaxEntries = 1 + ctx := context.Background() + + block0, err := EncodeBlockOption(SZX16, 0, true) + require.NoError(t, err) + + first := bw.cc.AcquireMessage(ctx) + first.SetCode(codes.POST) + first.SetToken(message.Token{0x01}) + first.SetOptionUint32(message.Block1, block0) + first.SetBody(bytes.NewReader(make([]byte, int(SZX16.Size())))) + w1 := responsewriter.New(bw.cc.AcquireMessage(ctx), bw.cc) + bw.Handle(w1, first, SZX16, uint32(SZX16.Size()), func(_ *responsewriter.ResponseWriter[*testClient], _ *pool.Message) { + require.Fail(t, "next must not be called while more blocks follow") + }) + require.Equal(t, codes.Continue, w1.Message().Code()) + + second := bw.cc.AcquireMessage(ctx) + second.SetCode(codes.POST) + second.SetToken(message.Token{0x02}) + second.SetOptionUint32(message.Block1, block0) + second.SetBody(bytes.NewReader(make([]byte, int(SZX16.Size())))) + w2 := responsewriter.New(bw.cc.AcquireMessage(ctx), bw.cc) + bw.Handle(w2, second, SZX16, uint32(SZX16.Size()), func(_ *responsewriter.ResponseWriter[*testClient], _ *pool.Message) { + require.Fail(t, "next must not be called when cache is full") + }) + require.Equal(t, codes.ServiceUnavailable, w2.Message().Code()) + require.Error(t, gotErr) + require.ErrorContains(t, gotErr, "receiving messages cache is full") + require.Len(t, bw.receivingMessagesCache.LoadAndDeleteAll(), 1) +} diff --git a/options/commonOptions.go b/options/commonOptions.go index dfad6344..fb395c00 100644 --- a/options/commonOptions.go +++ b/options/commonOptions.go @@ -543,6 +543,37 @@ func WithBlockwise(enable bool, szx blockwise.SZX, transferTimeout time.Duration } } +// BlockwiseReceivingMessagesCacheMaxEntriesOpt network option. +type BlockwiseReceivingMessagesCacheMaxEntriesOpt struct { + maxEntries int +} + +func (o BlockwiseReceivingMessagesCacheMaxEntriesOpt) UDPServerApply(cfg *udpServer.Config) { + cfg.BlockwiseReceivingMessagesCacheMaxEntries = o.maxEntries +} + +func (o BlockwiseReceivingMessagesCacheMaxEntriesOpt) DTLSServerApply(cfg *dtlsServer.Config) { + cfg.BlockwiseReceivingMessagesCacheMaxEntries = o.maxEntries +} + +func (o BlockwiseReceivingMessagesCacheMaxEntriesOpt) UDPClientApply(cfg *udpClient.Config) { + cfg.BlockwiseReceivingMessagesCacheMaxEntries = o.maxEntries +} + +func (o BlockwiseReceivingMessagesCacheMaxEntriesOpt) TCPServerApply(cfg *tcpServer.Config) { + cfg.BlockwiseReceivingMessagesCacheMaxEntries = o.maxEntries +} + +func (o BlockwiseReceivingMessagesCacheMaxEntriesOpt) TCPClientApply(cfg *tcpClient.Config) { + cfg.BlockwiseReceivingMessagesCacheMaxEntries = o.maxEntries +} + +// WithBlockwiseReceivingMessagesCacheMaxEntries configures max entries for +// in-flight incoming block-wise assemblies. Value 0 means unlimited. +func WithBlockwiseReceivingMessagesCacheMaxEntries(maxEntries int) BlockwiseReceivingMessagesCacheMaxEntriesOpt { + return BlockwiseReceivingMessagesCacheMaxEntriesOpt{maxEntries: maxEntries} +} + type OnNewConnFunc interface { tcpServer.OnNewConnFunc | udpServer.OnNewConnFunc } diff --git a/options/commonOptions_test.go b/options/commonOptions_test.go index b894183a..119f976c 100644 --- a/options/commonOptions_test.go +++ b/options/commonOptions_test.go @@ -59,6 +59,7 @@ func TestCommonTCPServerApply(t *testing.T) { options.WithInactivityMonitor(time.Minute, inactivityMonitor), options.WithPeriodicRunner(periodicRunner), options.WithBlockwise(true, blockwise.SZX16, time.Second), + options.WithBlockwiseReceivingMessagesCacheMaxEntries(0), options.WithOnNewConn(onNewConn), options.WithRequestMonitor(requestMonitor), options.WithMessagePool(mp), @@ -89,6 +90,7 @@ func TestCommonTCPServerApply(t *testing.T) { require.True(t, cfg.BlockwiseEnable) require.Equal(t, blockwise.SZX16, cfg.BlockwiseSZX) require.Equal(t, time.Second, cfg.BlockwiseTransferTimeout) + require.Equal(t, 0, cfg.BlockwiseReceivingMessagesCacheMaxEntries) // WithOnNewConn require.NotNil(t, cfg.OnNewConn) // WithRequestMonitor @@ -154,6 +156,7 @@ func TestCommonTCPClientApply(t *testing.T) { options.WithNetwork(network), options.WithPeriodicRunner(periodicRunner), options.WithBlockwise(true, blockwise.SZX16, time.Second), + options.WithBlockwiseReceivingMessagesCacheMaxEntries(0), options.WithCloseSocket(), options.WithDialer(dialer), options.WithMessagePool(mp), @@ -186,6 +189,7 @@ func TestCommonTCPClientApply(t *testing.T) { require.True(t, cfg.BlockwiseEnable) require.Equal(t, blockwise.SZX16, cfg.BlockwiseSZX) require.Equal(t, time.Second, cfg.BlockwiseTransferTimeout) + require.Equal(t, 0, cfg.BlockwiseReceivingMessagesCacheMaxEntries) // WithCloseSocket require.True(t, cfg.CloseSocket) // WithDialer @@ -254,6 +258,7 @@ func TestCommonUDPServerApply(t *testing.T) { options.WithInactivityMonitor(time.Minute, inactivityMonitor), options.WithPeriodicRunner(periodicRunner), options.WithBlockwise(true, blockwise.SZX16, time.Second), + options.WithBlockwiseReceivingMessagesCacheMaxEntries(0), options.WithOnNewConn(onNewConn), options.WithRequestMonitor(requestMonitor), options.WithMessagePool(mp), @@ -284,6 +289,7 @@ func TestCommonUDPServerApply(t *testing.T) { require.True(t, cfg.BlockwiseEnable) require.Equal(t, blockwise.SZX16, cfg.BlockwiseSZX) require.Equal(t, time.Second, cfg.BlockwiseTransferTimeout) + require.Equal(t, 0, cfg.BlockwiseReceivingMessagesCacheMaxEntries) // WithOnNewConn require.NotNil(t, cfg.OnNewConn) // WithRequestMonitor @@ -352,6 +358,7 @@ func TestCommonDTLSServerApply(t *testing.T) { options.WithInactivityMonitor(time.Minute, inactivityMonitor), options.WithPeriodicRunner(periodicRunner), options.WithBlockwise(true, blockwise.SZX16, time.Second), + options.WithBlockwiseReceivingMessagesCacheMaxEntries(0), options.WithOnNewConn(onNewConn), options.WithRequestMonitor(requestMonitor), options.WithMessagePool(mp), @@ -382,6 +389,7 @@ func TestCommonDTLSServerApply(t *testing.T) { require.True(t, cfg.BlockwiseEnable) require.Equal(t, blockwise.SZX16, cfg.BlockwiseSZX) require.Equal(t, time.Second, cfg.BlockwiseTransferTimeout) + require.Equal(t, 0, cfg.BlockwiseReceivingMessagesCacheMaxEntries) // WithOnNewConn require.NotNil(t, cfg.OnNewConn) // WithRequestMonitor @@ -448,6 +456,7 @@ func TestCommonUDPClientApply(t *testing.T) { options.WithNetwork(network), options.WithPeriodicRunner(periodicRunner), options.WithBlockwise(true, blockwise.SZX16, time.Second), + options.WithBlockwiseReceivingMessagesCacheMaxEntries(0), options.WithCloseSocket(), options.WithDialer(dialer), options.WithMessagePool(mp), @@ -480,6 +489,7 @@ func TestCommonUDPClientApply(t *testing.T) { require.True(t, cfg.BlockwiseEnable) require.Equal(t, blockwise.SZX16, cfg.BlockwiseSZX) require.Equal(t, time.Second, cfg.BlockwiseTransferTimeout) + require.Equal(t, 0, cfg.BlockwiseReceivingMessagesCacheMaxEntries) // WithCloseSocket require.True(t, cfg.CloseSocket) // WithDialer diff --git a/options/config/common.go b/options/config/common.go index e6fd876a..2df342bf 100644 --- a/options/config/common.go +++ b/options/config/common.go @@ -20,19 +20,20 @@ type ( ) type Common[C responsewriter.Client] struct { - LimitClientParallelRequests int64 - LimitClientEndpointParallelRequests int64 - Ctx context.Context - Errors ErrorFunc - PeriodicRunner periodic.Func - MessagePool *pool.Pool - GetToken client.GetTokenFunc - MaxMessageSize uint32 - BlockwiseTransferTimeout time.Duration - BlockwiseSZX blockwise.SZX - BlockwiseEnable bool - ProcessReceivedMessage ProcessReceivedMessageFunc[C] - ReceivedMessageQueueSize int + LimitClientParallelRequests int64 + LimitClientEndpointParallelRequests int64 + Ctx context.Context + Errors ErrorFunc + PeriodicRunner periodic.Func + MessagePool *pool.Pool + GetToken client.GetTokenFunc + MaxMessageSize uint32 + BlockwiseTransferTimeout time.Duration + BlockwiseReceivingMessagesCacheMaxEntries int + BlockwiseSZX blockwise.SZX + BlockwiseEnable bool + ProcessReceivedMessage ProcessReceivedMessageFunc[C] + ReceivedMessageQueueSize int } func NewCommon[C responsewriter.Client]() Common[C] { @@ -45,6 +46,7 @@ func NewCommon[C responsewriter.Client]() Common[C] { BlockwiseSZX: blockwise.SZX1024, BlockwiseEnable: true, BlockwiseTransferTimeout: time.Second * 3, + BlockwiseReceivingMessagesCacheMaxEntries: blockwise.DefaultReceivingMessagesCacheMaxEntries, PeriodicRunner: func(f func(now time.Time) bool) { go func() { for f(time.Now()) { diff --git a/tcp/client.go b/tcp/client.go index 5d1685e6..b9f19adb 100644 --- a/tcp/client.go +++ b/tcp/client.go @@ -30,7 +30,7 @@ func createBlockWiseFactory(cfg *client.Config) func(*client.Conn) *blockwise.Bl return func(cc *client.Conn) *blockwise.BlockWise[*client.Conn] { v := cc - return blockwise.New( + bw := blockwise.New( v, cfg.BlockwiseTransferTimeout, cfg.Errors, @@ -38,6 +38,8 @@ func createBlockWiseFactory(cfg *client.Config) func(*client.Conn) *blockwise.Bl return v.GetObservationRequest(token) }, ) + bw.SetReceivingMessagesCacheMaxEntries(cfg.BlockwiseReceivingMessagesCacheMaxEntries) + return bw } } diff --git a/tcp/server/server.go b/tcp/server/server.go index 89800c8c..7e0f5fe2 100644 --- a/tcp/server/server.go +++ b/tcp/server/server.go @@ -188,7 +188,7 @@ func (s *Server) createConn(connection *coapNet.Conn, inactivityMonitor client.I } if s.cfg.BlockwiseEnable { createBlockWise = func(cc *client.Conn) *blockwise.BlockWise[*client.Conn] { - return blockwise.New( + bw := blockwise.New( cc, s.cfg.BlockwiseTransferTimeout, s.cfg.Errors, @@ -196,6 +196,8 @@ func (s *Server) createConn(connection *coapNet.Conn, inactivityMonitor client.I return nil, false }, ) + bw.SetReceivingMessagesCacheMaxEntries(s.cfg.BlockwiseReceivingMessagesCacheMaxEntries) + return bw } } cfg := client.DefaultConfig diff --git a/udp/blockwise_test.go b/udp/blockwise_test.go index a1ac7e16..b07f0bc4 100644 --- a/udp/blockwise_test.go +++ b/udp/blockwise_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "io" + "net" "sync" "testing" "time" @@ -22,6 +23,81 @@ import ( "go.uber.org/atomic" ) +type rawOption struct { + id uint16 + val []byte +} + +func rawUintValue(v uint32) []byte { + switch { + case v == 0: + return nil + case v < 256: + return []byte{byte(v)} + case v < 65536: + return []byte{byte(v >> 8), byte(v)} + default: + return []byte{byte(v >> 24), byte(v >> 16), byte(v >> 8), byte(v)} + } +} + +func rawBlockOption(num uint32, more bool, szx uint8) []byte { + moreBit := uint32(0) + if more { + moreBit = 1 + } + return rawUintValue((num << 4) | (moreBit << 3) | uint32(szx)) +} + +func encodeRawHeader(delta, length uint16) []byte { + first := byte(0) + var ext []byte + switch { + case delta < 13: + first |= byte(delta << 4) + case delta < 269: + first |= 13 << 4 + ext = append(ext, byte(delta-13)) + default: + first |= 14 << 4 + ext = append(ext, byte((delta-269)>>8), byte(delta-269)) + } + switch { + case length < 13: + first |= byte(length) + case length < 269: + first |= 13 + ext = append(ext, byte(length-13)) + default: + first |= 14 + ext = append(ext, byte((length-269)>>8), byte(length-269)) + } + return append([]byte{first}, ext...) +} + +func encodeRawOptions(opts []rawOption) []byte { + var b bytes.Buffer + prev := uint16(0) + for _, opt := range opts { + b.Write(encodeRawHeader(opt.id-prev, uint16(len(opt.val)))) + b.Write(opt.val) + prev = opt.id + } + return b.Bytes() +} + +func rawMessage(typ, code uint8, messageID uint16, token []byte, opts []rawOption, payload []byte) []byte { + firstByte := byte(1<<6) | (typ << 4) | uint8(len(token)) + out := []byte{firstByte, code, byte(messageID >> 8), byte(messageID)} + out = append(out, token...) + out = append(out, encodeRawOptions(opts)...) + if len(payload) > 0 { + out = append(out, 0xFF) + out = append(out, payload...) + } + return out +} + // TestServerBlockwiseLargeResponse is an end-to-end regression test for the Windows // WriteMsgUDP n=0 bug. A response larger than a single block forces a Block2 transfer. // @@ -99,3 +175,69 @@ func TestServerBlockwiseLargeResponse(t *testing.T) { require.GreaterOrEqual(t, calls, int32(1)) require.LessOrEqual(t, calls, int32(3)) } + +func TestServerBlockwiseMissingPreviousBlockReturnsRequestEntityIncomplete(t *testing.T) { + listener, err := coapNet.NewListenUDP("udp4", "127.0.0.1:0") + require.NoError(t, err) + defer func() { + errClose := listener.Close() + require.NoError(t, errClose) + }() + + server := udp.NewServer( + options.WithBlockwise(true, blockwise.SZX1024, time.Second*5), + options.WithHandlerFunc(func(w *responsewriter.ResponseWriter[*client.Conn], _ *pool.Message) { + errSet := w.SetResponse(codes.Changed, message.TextPlain, bytes.NewReader([]byte("ok"))) + require.NoError(t, errSet) + }), + ) + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + errServe := server.Serve(listener) + assert.NoError(t, errServe) + }() + defer func() { + server.Stop() + wg.Wait() + }() + + conn, err := net.DialUDP("udp4", nil, listener.LocalAddr().(*net.UDPAddr)) + require.NoError(t, err) + defer func() { + errClose := conn.Close() + require.NoError(t, errClose) + }() + + roundTrip := func(packet []byte, timeout time.Duration) []byte { + _, errWrite := conn.Write(packet) + require.NoError(t, errWrite) + errDeadline := conn.SetReadDeadline(time.Now().Add(timeout)) + require.NoError(t, errDeadline) + buf := make([]byte, 1500) + n, errRead := conn.Read(buf) + require.NoError(t, errRead) + return buf[:n] + } + + token := []byte{0x09} + firstBlock := rawMessage(1, uint8(codes.POST), 1, token, []rawOption{ + {id: uint16(message.URIPath), val: []byte("upload")}, + {id: uint16(message.Block1), val: rawBlockOption(0, true, 0)}, + {id: uint16(message.Size1), val: rawUintValue(32)}, + }, bytes.Repeat([]byte("A"), 16)) + firstResponse := roundTrip(firstBlock, time.Second) + require.GreaterOrEqual(t, len(firstResponse), 2) + require.Equal(t, byte(codes.Continue), firstResponse[1]) + + finalBlock := rawMessage(1, uint8(codes.POST), 2, token, []rawOption{ + {id: uint16(message.URIPath), val: []byte("upload")}, + {id: uint16(message.Block1), val: rawBlockOption(2, false, 0)}, + {id: uint16(message.Size1), val: rawUintValue(32)}, + }, bytes.Repeat([]byte("C"), 16)) + finalResponse := roundTrip(finalBlock, 2*time.Second) + require.GreaterOrEqual(t, len(finalResponse), 2) + require.Equal(t, byte(codes.RequestEntityIncomplete), finalResponse[1]) +} diff --git a/udp/client.go b/udp/client.go index 4a68b358..57295722 100644 --- a/udp/client.go +++ b/udp/client.go @@ -74,7 +74,7 @@ func Client(conn *net.UDPConn, opts ...Option) *client.Conn { if cfg.BlockwiseEnable { createBlockWise = func(cc *client.Conn) *blockwise.BlockWise[*client.Conn] { v := cc - return blockwise.New( + bw := blockwise.New( v, cfg.BlockwiseTransferTimeout, cfg.Errors, @@ -82,6 +82,8 @@ func Client(conn *net.UDPConn, opts ...Option) *client.Conn { return v.GetObservationRequest(token) }, ) + bw.SetReceivingMessagesCacheMaxEntries(cfg.BlockwiseReceivingMessagesCacheMaxEntries) + return bw } } diff --git a/udp/server/server.go b/udp/server/server.go index 8fab3b45..c9dfd898 100644 --- a/udp/server/server.go +++ b/udp/server/server.go @@ -331,7 +331,7 @@ func (s *Server) getOrCreateConn(udpConn *coapNet.UDPConn, raddr *net.UDPAddr, l if s.cfg.BlockwiseEnable { createBlockWise = func(cc *client.Conn) *blockwise.BlockWise[*client.Conn] { v := cc - return blockwise.New( + bw := blockwise.New( v, s.cfg.BlockwiseTransferTimeout, s.cfg.Errors, @@ -348,7 +348,10 @@ func (s *Server) getOrCreateConn(udpConn *coapNet.UDPConn, raddr *net.UDPAddr, l msg.SetMessageID(m.MessageID()) return msg }) - }) + }, + ) + bw.SetReceivingMessagesCacheMaxEntries(s.cfg.BlockwiseReceivingMessagesCacheMaxEntries) + return bw } } session := NewSession(