Repository: nextstrain/ncov
Issue: `/dev/stdout` not handled properly for the snakemake workflow

**Current Behavior**
When I run the workflow with today's (2021-10-07) packages (`sequences_fasta_2021-10-07.tar.xz` and `metadata_tsv_2021-10-07.tar.xz`), I run into an error with the `align` rule.

**Expected behavior**
The `align` rule should pass without flagrant errors.

**How to reproduce**
Steps to reproduce the current behavior:
1. Download `sequences_fasta_2021-10-07.tar.xz` and `metadata_tsv_2021-10-07.tar.xz` from GISAID.
2. Place them into position as data sources for your workflow's profile.
3. Run the workflow with the latest commit on the ~~`main`~~ `master` branch.
4. See the error:

```blank
Error in rule align:
    jobid: 181
    output: results/aligned_gisaid.fasta.xz, results/insertions_gisaid.tsv, results/translations/seqs_gisaid.gene.ORF1a.fasta.xz, results/translations/seqs_gisaid.gene.ORF1b.fasta.xz, results/translations/seqs_gisaid.gene.S.fasta.xz, results/translations/seqs_gisaid.gene.ORF3a.fasta.xz, results/translations/seqs_gisaid.gene.E.fasta.xz, results/translations/seqs_gisaid.gene.M.fasta.xz, results/translations/seqs_gisaid.gene.ORF6.fasta.xz, results/translations/seqs_gisaid.gene.ORF7a.fasta.xz, results/translations/seqs_gisaid.gene.ORF7b.fasta.xz, results/translations/seqs_gisaid.gene.ORF8.fasta.xz, results/translations/seqs_gisaid.gene.N.fasta.xz, results/translations/seqs_gisaid.gene.ORF9b.fasta.xz    
    log: logs/align_gisaid.txt (check log file(s) for error message)
    shell:
                                                                                                                                                                                 
        python3 scripts/sanitize_sequences.py             --sequences data/sequences_fasta_2021-10-07.tar.xz             --strip-prefixes hCoV-19/ SARS-CoV-2/             --output /dev/stdout 2> logs/sanitize_sequences_gisaid.txt             | nextalign             --jobs=8             --reference defaults/reference_seq.fasta             --genemap defaults/annotation.gff             --genes ORF1a,ORF1b,S,ORF3a,E,M,ORF6,ORF7a,ORF7b,ORF8,N,ORF9b             --sequences /dev/stdin             --output-dir results/translations             --output-basename seqs_gisaid             --output-fasta results/aligned_gisaid.fasta             --output-insertions results/insertions_gisaid.tsv > logs/align_gisaid.txt 2>&1;                                                                                                                                                                       
        xz -2 results/aligned_gisaid.fasta;                                                                                                                                      
        xz -2 results/translations/seqs_gisaid*.fasta                                                                                                                            
                                                                                                                                                                                 
        (one of the commands exited with non-zero exit code; note that snakemake uses bash strict mode!)                                                                         
Logfile logs/align_gisaid.txt:
[ERROR] Nextalign: Error: when running the internal parallel pipeline: When parsing input sequences: Input stream ("/dev/stdin") is empty or corrupted. Aborting.
```

**Possible solution**
(optional)

**Your environment: if browsing Nextstrain online**
 - Operating system:
 - Browser:

**Your environment: if running Nextstrain locally**
 - Operating system: Debian Buster
 - Browser:
 - Version (e.g. `auspice 2.7.0`): 

**Additional context**
Add any other context about the problem here.

I looked in `logs/sanitize_sequences_gisaid.txt`, and it gave me this output:

```none
Traceback (most recent call last):
  File "scripts/sanitize_sequences.py", line 102, in <module>
    "sequences"
  File "/nextstrain/build/scripts/utils.py", line 62, in extract_tar_file_contents
    members=(internal_member,)
  File "/usr/local/lib/python3.7/tarfile.py", line 2002, in extractall
    numeric_owner=numeric_owner)
  File "/usr/local/lib/python3.7/tarfile.py", line 2044, in extract
    numeric_owner=numeric_owner)
  File "/usr/local/lib/python3.7/tarfile.py", line 2114, in _extract_member
    self.makefile(tarinfo, targetpath)
  File "/usr/local/lib/python3.7/tarfile.py", line 2163, in makefile
    copyfileobj(source, target, tarinfo.size, ReadError, bufsize)
  File "/usr/local/lib/python3.7/tarfile.py", line 250, in copyfileobj
    dst.write(buf)
OSError: [Errno 28] No space left on device
```

Unfortunately, I can't diagnose this further until tomorrow.

Changed files:
scripts/sanitize_sequences.py, scripts/utils.py

diff --git a/scripts/sanitize_sequences.py b/scripts/sanitize_sequences.py
index 15ca6bd3..36097cb9 100644
--- a/scripts/sanitize_sequences.py
+++ b/scripts/sanitize_sequences.py
@@ -5,7 +5,7 @@ from pathlib import Path
 import re
 import sys
 
-from utils import extract_tar_file_contents
+from utils import stream_tar_file_contents
 
 
 class DuplicateSequenceError(ValueError):
@@ -91,18 +91,18 @@ if __name__ == '__main__':
     args = parser.parse_args()
 
     sequence_files = []
-    temporary_dirs = []
+    tar_files = []
     for sequence_filename in args.sequences:
         # If the input is a tarball, try to find a sequence file inside the
         # archive.
         if ".tar" in Path(sequence_filename).suffixes:
             try:
-                temporary_dir, sequence_file = extract_tar_file_contents(
+                sequence_file, tar_file = stream_tar_file_contents(
                     sequence_filename,
                     "sequences"
                 )
                 sequence_files.append(sequence_file)
-                temporary_dirs.append(temporary_dir)
+                tar_files.append(tar_file)
             except FileNotFoundError as error:
                 print(f"ERROR: {error}", file=sys.stderr)
                 sys.exit(1)
@@ -136,6 +136,5 @@ if __name__ == '__main__':
             sys.exit(1)
 
     # Clean up temporary directory and files that came from a tarball.
-    for temporary_dir in temporary_dirs:
-        print(f"Cleaning up temporary files in {temporary_dir.name}", file=sys.stderr)
-        temporary_dir.cleanup()
+    for tar_file in tar_files:
+        tar_file.close()
diff --git a/scripts/utils.py b/scripts/utils.py
index 1060bb69..ee6d3154 100644
--- a/scripts/utils.py
+++ b/scripts/utils.py
@@ -1,3 +1,4 @@
+import lzma
 from pathlib import Path
 import sys
 import tarfile
@@ -71,3 +72,59 @@ def extract_tar_file_contents(filename, filetype):
     # the temporary directory when its object is destroyed. For more details, see
     # https://docs.python.org/3/library/tempfile.html#tempfile.TemporaryDirectory
     return temporary_dir, extracted_file_path
+
+
+def stream_tar_file_contents(filename, filetype):
+    """Try to extract the contents of a given file type (e.g., metadata or
+    sequences) from the given tar filename.
+
+    Parameters
+    ----------
+    filename : str or Path-like
+        Path to the tar archive to search for the given file type.
+    filetype : str
+        Type of file to search for in the given tar archive based on the
+        associated file extension.
+
+    Returns
+    -------
+    io.BufferedReader :
+        A stream of the requested file from the tar archive.
+    TarFile :
+        A handle to the original tar archive to be closed when the stream has
+        been read.
+
+    Raises
+    ------
+    FileNotFoundError :
+        When a file with the the requested file type's extension could not be
+        found in the given tar archive.
+
+    """
+    extension = EXTENSION_BY_FILETYPE[filetype]
+
+    tar = tarfile.open(filename)
+    internal_file = None
+    for member in tar.getmembers():
+        suffixes = Path(member.name).suffixes
+
+        if extension in suffixes:
+            # By default, return the binary stream for the member file.
+            internal_file = tar.extractfile(member.name)
+
+            if ".xz" in suffixes:
+                # Check for LZMA-compressed data and open these with the
+                # corresponding library.
+                internal_file = lzma.open(internal_file, "rt")
+            elif extension == ".fasta":
+                # For sequence data, handle decoding of the binary stream prior
+                # to passing the data back to the caller.
+                internal_file = (line.decode("utf-8") for line in internal_file)
+
+            break
+
+    if internal_file is None:
+        tar.close()
+        raise FileNotFoundError(f"Could not find a {filetype} file in '{filename}'")
+
+    return internal_file, tar