[FIXED] Object store context put and get issues (#1260)

Signed-off-by: Piotr Piotrowski <piotr@synadia.com>

object.go

@@ -368,13 +368,19 @@ func (obs *obs) Put(meta *ObjectMeta, r io.Reader, opts ...ObjectOpt) (*ObjectIn
 		return perr
 	}
 
-	purgePartial := func() { obs.js.purgeStream(obs.stream, &StreamPurgeRequest{Subject: chunkSubj}) }
-
 	// Create our own JS context to handle errors etc.
 	js, err := obs.js.nc.JetStream(PublishAsyncErrHandler(func(js JetStream, _ *Msg, err error) { setErr(err) }))
 	if err != nil {
 		return nil, err
 	}
+	purgePartial := func() {
+		// wait until all pubs are complete or up to default timeout before attempting purge
+		select {
+		case <-js.PublishAsyncComplete():
+		case <-time.After(obs.js.opts.wait):
+		}
+		obs.js.purgeStream(obs.stream, &StreamPurgeRequest{Subject: chunkSubj})
+	}
 
 	m, h := NewMsg(chunkSubj), sha256.New()
 	chunk, sent, total := make([]byte, meta.Opts.ChunkSize), 0, uint64(0)
@@ -1207,7 +1213,7 @@ func (o *objResult) Read(p []byte) (n int, err error) {
 		}
 	}
 	if o.err != nil {
-		return 0, err
+		return 0, o.err
 	}
 	if o.r == nil {
 		return 0, io.EOF

test/norace_test.go

@@ -59,9 +59,8 @@ func TestNoRaceObjectContextOpt(t *testing.T) {
 	expectOk(t, err)
 
 	ctx, cancel = context.WithTimeout(context.Background(), 2*time.Second)
-	time.AfterFunc(100*time.Millisecond, cancel)
+	time.AfterFunc(10*time.Millisecond, cancel)
 
-	time.AfterFunc(20*time.Millisecond, func() { shutdownJSServerAndRemoveStorage(t, s) })
 	start = time.Now()
 	_, err = obs.GetBytes("BLOB", nats.Context(ctx))
 	expectErr(t, err)