Merge pull request #2818 from adri326/stream-cleanup

Fix various issues around close/1 and stream realiasing

src/machine/loader.rs

@@ -1808,7 +1808,7 @@ impl Machine {
     pub(crate) fn push_load_context(&mut self) -> CallResult {
         let stream = self.machine_st.get_stream_or_alias(
             self.machine_st.registers[1],
-            &self.indices.stream_aliases,
+            &self.indices,
             atom!("$push_load_context"),
             2,
         )?;

src/machine/machine_indices.rs

@@ -7,8 +7,9 @@ use crate::atom_table::*;
 use crate::forms::*;
 use crate::machine::loader::*;
 use crate::machine::machine_state::*;
-use crate::machine::streams::Stream;
+use crate::machine::streams::{Stream, StreamOptions};
 use crate::machine::ClauseType;
+use crate::machine::MachineStubGen;
 
 use fxhash::FxBuildHasher;
 use indexmap::{IndexMap, IndexSet};
@@ -261,8 +262,8 @@ pub struct IndexStore {
     pub(super) meta_predicates: MetaPredicateDir,
     pub(super) modules: ModuleDir,
     pub(super) op_dir: OpDir,
-    pub(super) streams: StreamDir,
+    streams: StreamDir,
-    pub(super) stream_aliases: StreamAliasDir,
+    stream_aliases: StreamAliasDir,
 }
 
 impl IndexStore {
@@ -459,6 +460,113 @@ impl IndexStore {
         }
     }
 
+    pub(crate) fn add_stream(
+        &mut self,
+        stream: Stream,
+        stub_name: Atom,
+        stub_arity: usize,
+    ) -> Result<(), MachineStubGen> {
+        if let Some(alias) = stream.options().get_alias() {
+            if self.stream_aliases.contains_key(&alias) {
+                return Err(Box::new(move |machine_st| {
+                    machine_st.occupied_alias_permission_error(alias, stub_name, stub_arity)
+                }));
+            }
+
+            self.stream_aliases.insert(alias, stream);
+        }
+
+        self.streams.insert(stream);
+
+        Ok(())
+    }
+
+    pub(crate) fn remove_stream(&mut self, stream: Stream) {
+        if let Some(alias) = stream.options().get_alias() {
+            debug_assert_eq!(self.stream_aliases.get(&alias), Some(&stream));
+            assert!(!is_protected_alias(alias));
+
+            self.stream_aliases.swap_remove(&alias);
+        }
+        self.streams.remove(&stream);
+    }
+
+    pub(crate) fn update_stream_options<F: Fn(&mut StreamOptions)>(
+        &mut self,
+        mut stream: Stream,
+        callback: F,
+    ) {
+        if let Some(prev_alias) = stream.options().get_alias() {
+            debug_assert_eq!(self.stream_aliases.get(&prev_alias), Some(&stream));
+        }
+        let options = stream.options_mut();
+        let prev_alias = options.get_alias();
+
+        callback(options);
+
+        if options.get_alias() != prev_alias {
+            if prev_alias.map(is_protected_alias).unwrap_or(false)
+                || options
+                    .get_alias()
+                    .map(|alias| self.has_stream(alias))
+                    .unwrap_or(false)
+            {
+                // user_input, user_output and user_error cannot be realiased,
+                // and realiasing cannot shadow an existing stream.
+                options.set_alias_to_atom_opt(prev_alias);
+                return;
+            }
+
+            if let Some(prev_alias) = prev_alias {
+                self.stream_aliases.swap_remove(&prev_alias);
+            }
+            if let Some(new_alias) = options.get_alias() {
+                self.stream_aliases.insert(new_alias, stream);
+            }
+        }
+    }
+
+    pub(crate) fn has_stream(&self, alias: Atom) -> bool {
+        self.stream_aliases.contains_key(&alias)
+    }
+
+    /// ## Warning
+    ///
+    /// The returned stream's options should only be modified through
+    /// [`IndexStore::update_stream_options`], to avoid breaking the
+    /// invariants of [`IndexStore`].
+    pub(crate) fn get_stream(&self, alias: Atom) -> Option<Stream> {
+        self.stream_aliases.get(&alias).copied()
+    }
+
+    pub(crate) fn iter_streams<'a, R: std::ops::RangeBounds<Stream>>(
+        &'a self,
+        range: R,
+    ) -> impl Iterator<Item = Stream> + 'a {
+        self.streams.range(range).into_iter().copied()
+    }
+
+    /// Forcibly sets `alias` to `stream`.
+    /// If there was a previous stream with that alias, it will lose that alias.
+    ///
+    /// Consider using [`add_stream`](Self::add_stream) if you wish to instead
+    /// return an error when stream aliases conflict.
+    pub(crate) fn set_stream(&mut self, alias: Atom, mut stream: Stream) {
+        if let Some(mut prev_stream) = self.get_stream(alias) {
+            if prev_stream == stream {
+                // Nothing to do, as the stream is already present
+                return;
+            }
+
+            prev_stream.options_mut().set_alias_to_atom_opt(None);
+        }
+
+        stream.options_mut().set_alias_to_atom_opt(Some(alias));
+
+        self.stream_aliases.insert(alias, stream);
+        self.streams.insert(stream);
+    }
+
     #[inline]
     pub(super) fn new() -> Self {
         index_store!(
@@ -468,3 +576,13 @@ impl IndexStore {
         )
     }
 }
+
+/// A stream is said to have a "protected" alias if modifying its
+/// alias would cause breakage in other parts of the code.
+///
+/// A stream with a protected alias cannot be realiased through
+/// [`IndexStore::update_stream_options`]. Instead, one has to use
+/// [`IndexStore::set_stream`] to do so.
+fn is_protected_alias(alias: Atom) -> bool {
+    alias == atom!("user_input") || alias == atom!("user_output") || alias == atom!("user_error")
+}

src/machine/mod.rs

@@ -483,32 +483,16 @@ impl Machine {
         }
     }
 
+    /// Ensures that [`Machine::indices`] properly reflects
+    /// the streams stored in [`Machine::user_input`], [`Machine::user_output`]
+    /// and [`Machine::user_error`].
     pub(crate) fn configure_streams(&mut self) {
-        self.user_input
-            .options_mut()
-            .set_alias_to_atom_opt(Some(atom!("user_input")));
-
         self.indices
-            .stream_aliases
+            .set_stream(atom!("user_input"), self.user_input);
-            .insert(atom!("user_input"), self.user_input);
-
-        self.indices.streams.insert(self.user_input);
-
-        self.user_output
-            .options_mut()
-            .set_alias_to_atom_opt(Some(atom!("user_output")));
-
         self.indices
-            .stream_aliases
+            .set_stream(atom!("user_output"), self.user_output);
-            .insert(atom!("user_output"), self.user_output);
-
-        self.indices.streams.insert(self.user_output);
-
         self.indices
-            .stream_aliases
+            .set_stream(atom!("user_error"), self.user_error);
-            .insert(atom!("user_error"), self.user_error);
-
-        self.indices.streams.insert(self.user_error);
     }
 
     #[inline(always)]

src/machine/streams.rs

@@ -640,7 +640,7 @@ impl Stream {
         }
     }
 
-    pub fn options_mut(&mut self) -> &mut StreamOptions {
+    pub(super) fn options_mut(&mut self) -> &mut StreamOptions {
         match self {
             Stream::Byte(ref mut ptr) => &mut ptr.options,
             Stream::InputFile(ref mut ptr) => &mut ptr.options,
@@ -1288,11 +1288,10 @@ impl Stream {
         ))
     }
 
+    /// Drops the stream handle and marks the arena pointer as [`ArenaHeaderTag::Dropped`].
     #[inline]
     pub(crate) fn close(&mut self) -> Result<(), std::io::Error> {
-        let mut stream = std::mem::replace(self, Stream::Null(StreamOptions::default()));
+        match self {
-
-        match stream {
             Stream::NamedTcp(ref mut tcp_stream) => {
                 tcp_stream.inner_mut().tcp_stream.shutdown(Shutdown::Both)
             }
@@ -1322,7 +1321,20 @@ impl Stream {
 
                 Ok(())
             }
-            _ => Ok(()),
+            Stream::Byte(mut stream) => {
+                stream.drop_payload();
+                Ok(())
+            }
+            Stream::StaticString(mut stream) => {
+                stream.drop_payload();
+                Ok(())
+            }
+
+            Stream::Null(_) => Ok(()),
+
+            Stream::Readline(_) | Stream::StandardOutput(_) | Stream::StandardError(_) => {
+                unreachable!();
+            }
         }
     }
 
@@ -1464,6 +1476,10 @@ impl MachineState {
         }
     }
 
+    /// ## Warning
+    ///
+    /// The options of streams stored in `Machine::indices` should only
+    /// be modified through [`IndexStore::update_stream_options`].
     pub(crate) fn get_stream_options(
         &mut self,
         alias: HeapCellValue,
@@ -1579,10 +1595,23 @@ impl MachineState {
         options
     }
 
+    /// If `addr` is a [`Cons`](HeapCellValueTag::Cons) to a stream, then returns it.
+    ///
+    /// If it is an atom or a string, then this searches for the corresponding stream
+    /// inside of [`self.indices`], returning it.
+    ///
+    /// ## Warning
+    ///
+    /// **Do not directly modify [`stream.options_mut()`](Stream::options_mut)
+    /// on the returned stream.**
+    ///
+    /// Other functions rely on the invariants of [`IndexStore`], which may
+    /// become invalidated by the direct modification of a stream's option (namely,
+    /// its alias name). Instead, use [`IndexStore::update_stream_options`].
     pub(crate) fn get_stream_or_alias(
         &mut self,
         addr: HeapCellValue,
-        stream_aliases: &StreamAliasDir,
+        indices: &IndexStore,
         caller: Atom,
         arity: usize,
     ) -> Result<Stream, MachineStub> {
@@ -1592,8 +1621,8 @@ impl MachineState {
                         (HeapCellValueTag::Atom, (name, arity)) => {
                             debug_assert_eq!(arity, 0);
 
-                            return match stream_aliases.get(&name) {
+                            return match indices.get_stream(name) {
-                                Some(stream) if !stream.is_null_stream() => Ok(*stream),
+                                Some(stream) if !stream.is_null_stream() => Ok(stream),
                                 _ => {
                                     let stub = functor_stub(caller, arity);
                                     let addr = atom_as_cell!(name);
@@ -1610,8 +1639,8 @@ impl MachineState {
 
                             debug_assert_eq!(arity, 0);
 
-                            return match stream_aliases.get(&name) {
+                            return match indices.get_stream(name) {
-                                Some(stream) if !stream.is_null_stream() => Ok(*stream),
+                                Some(stream) if !stream.is_null_stream() => Ok(stream),
                                 _ => {
                                     let stub = functor_stub(caller, arity);
                                     let addr = atom_as_cell!(name);
@@ -1801,7 +1830,7 @@ impl MachineState {
 
         // 8.11.5.3l)
         if let Some(alias) = options.get_alias() {
-            if indices.stream_aliases.contains_key(&alias) {
+            if indices.has_stream(alias) {
                 return Err(self.occupied_alias_permission_error(alias, atom!("open"), 4));
             }
         }
@@ -1893,3 +1922,91 @@ impl MachineState {
         }
     }
 }
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use crate::machine::config::*;
+
+    #[test]
+    #[cfg_attr(miri, ignore)]
+    fn close_memory_user_output_stream() {
+        let mut machine = MachineBuilder::new()
+            .with_streams(StreamConfig::in_memory())
+            .build();
+
+        let results = machine
+            .run_query(
+                "\\+ \\+ (current_output(Stream), close(Stream)), write(user_output, hello).",
+            )
+            .collect::<Vec<_>>();
+
+        assert_eq!(results.len(), 1);
+        assert!(results[0].is_ok());
+
+        let mut actual = String::new();
+        machine.user_output.read_to_string(&mut actual).unwrap();
+        assert_eq!(actual, "hello");
+    }
+
+    #[test]
+    #[cfg_attr(miri, ignore)]
+    fn close_memory_user_output_stream_twice() {
+        let mut machine = MachineBuilder::new()
+            .with_streams(StreamConfig::in_memory())
+            .build();
+
+        let results = machine
+            .run_query("\\+ \\+ (current_output(Stream), close(Stream), close(Stream)).")
+            .collect::<Vec<_>>();
+
+        assert_eq!(results.len(), 1);
+        assert!(results[0].is_ok());
+    }
+
+    #[test]
+    #[cfg_attr(miri, ignore)]
+    fn close_realiased_stream() {
+        let mut machine = MachineBuilder::new().build();
+
+        let results = machine
+            .run_query(
+                r#"
+                \+ \+ (
+                    open("README.md", read, S, [alias(readme)]),
+                    open(stream(S), read, _, [alias(another_alias)]),
+                    close(S)
+                ),
+                open("README.md", read, _, [alias(readme)]).
+            "#,
+            )
+            .collect::<Vec<_>>();
+
+        assert_eq!(results.len(), 1);
+        assert!(results[0].is_ok());
+    }
+
+    #[test]
+    #[cfg_attr(miri, ignore)]
+    fn close_realiased_user_output() {
+        let mut machine = MachineBuilder::new()
+            .with_streams(StreamConfig::in_memory())
+            .build();
+
+        let results = machine
+            .run_query(
+                r#"
+                \+ \+ (
+                    open("README.md", read, S),
+                    open(stream(S), read, _, [alias(user_output)]),
+                    close(S)
+                ),
+                write(user_output, hello).
+            "#,
+            )
+            .collect::<Vec<_>>();
+
+        assert_eq!(results.len(), 1);
+        assert!(results[0].is_ok());
+    }
+}

src/machine/system_calls.rs

@@ -41,7 +41,6 @@ use indexmap::IndexSet;
 
 use std::cell::Cell;
 use std::cmp::Ordering;
-use std::collections::BTreeSet;
 use std::convert::TryFrom;
 use std::env;
 #[cfg(feature = "ffi")]
@@ -55,7 +54,6 @@ use std::mem;
 use std::net::{SocketAddr, ToSocketAddrs};
 use std::net::{TcpListener, TcpStream};
 use std::num::NonZeroU32;
-use std::ops::Sub;
 use std::process;
 #[cfg(feature = "http")]
 use std::str::FromStr;
@@ -2543,7 +2541,7 @@ impl Machine {
 
         let mut stream = self.machine_st.get_stream_or_alias(
             self.machine_st.registers[1],
-            &self.indices.stream_aliases,
+            &self.indices,
             atom!("peek_byte"),
             2,
         )?;
@@ -2634,7 +2632,7 @@ impl Machine {
 
         let mut stream = self.machine_st.get_stream_or_alias(
             self.machine_st.registers[1],
-            &self.indices.stream_aliases,
+            &self.indices,
             atom!("peek_char"),
             2,
         )?;
@@ -2726,7 +2724,7 @@ impl Machine {
 
         let mut stream = self.machine_st.get_stream_or_alias(
             self.machine_st.registers[1],
-            &self.indices.stream_aliases,
+            &self.indices,
             atom!("peek_code"),
             2,
         )?;
@@ -3147,7 +3145,7 @@ impl Machine {
     pub(crate) fn put_code(&mut self) -> CallResult {
         let mut stream = self.machine_st.get_stream_or_alias(
             self.machine_st.registers[1],
-            &self.indices.stream_aliases,
+            &self.indices,
             atom!("put_code"),
             2,
         )?;
@@ -3199,7 +3197,7 @@ impl Machine {
     pub(crate) fn put_char(&mut self) -> CallResult {
         let mut stream = self.machine_st.get_stream_or_alias(
             self.machine_st.registers[1],
-            &self.indices.stream_aliases,
+            &self.indices,
             atom!("put_char"),
             2,
         )?;
@@ -3243,7 +3241,7 @@ impl Machine {
     pub(crate) fn put_chars(&mut self) -> CallResult {
         let mut stream = self.machine_st.get_stream_or_alias(
             self.machine_st.registers[1],
-            &self.indices.stream_aliases,
+            &self.indices,
             atom!("$put_chars"),
             2,
         )?;
@@ -3292,7 +3290,7 @@ impl Machine {
     pub(crate) fn put_byte(&mut self) -> CallResult {
         let mut stream = self.machine_st.get_stream_or_alias(
             self.machine_st.registers[1],
-            &self.indices.stream_aliases,
+            &self.indices,
             atom!("put_byte"),
             2,
         )?;
@@ -3359,7 +3357,7 @@ impl Machine {
     pub(crate) fn get_byte(&mut self) -> CallResult {
         let mut stream = self.machine_st.get_stream_or_alias(
             self.machine_st.registers[1],
-            &self.indices.stream_aliases,
+            &self.indices,
             atom!("get_byte"),
             2,
         )?;
@@ -3444,7 +3442,7 @@ impl Machine {
     pub(crate) fn get_char(&mut self) -> CallResult {
         let mut stream = self.machine_st.get_stream_or_alias(
             self.machine_st.registers[1],
-            &self.indices.stream_aliases,
+            &self.indices,
             atom!("get_char"),
             2,
         )?;
@@ -3539,7 +3537,7 @@ impl Machine {
     pub(crate) fn get_n_chars(&mut self) -> CallResult {
         let stream = self.machine_st.get_stream_or_alias(
             self.machine_st.registers[1],
-            &self.indices.stream_aliases,
+            &self.indices,
             atom!("get_n_chars"),
             3,
         )?;
@@ -3608,7 +3606,7 @@ impl Machine {
     pub(crate) fn get_code(&mut self) -> CallResult {
         let mut stream = self.machine_st.get_stream_or_alias(
             self.machine_st.registers[1],
-            &self.indices.stream_aliases,
+            &self.indices,
             atom!("get_code"),
             2,
         )?;
@@ -3715,19 +3713,11 @@ impl Machine {
 
     #[inline(always)]
     pub(crate) fn first_stream(&mut self) {
-        let mut first_stream = None;
+        let first_stream = self
-        let mut null_streams = BTreeSet::new();
+            .indices
-
+            .iter_streams(..)
-        for stream in self.indices.streams.iter().cloned() {
+            .filter(|s| !s.is_null_stream())
-            if !stream.is_null_stream() {
+            .next();
-                first_stream = Some(stream);
-                break;
-            } else {
-                null_streams.insert(stream);
-            }
-        }
-
-        self.indices.streams = self.indices.streams.sub(&null_streams);
 
         if let Some(first_stream) = first_stream {
             let stream = stream_as_cell!(first_stream);
@@ -3743,20 +3733,12 @@ impl Machine {
     #[inline(always)]
     pub(crate) fn next_stream(&mut self) {
         let prev_stream = cell_as_stream!(self.deref_register(1));
-
+        let next_stream = self
-        let mut next_stream = None;
+            .indices
-        let mut null_streams = BTreeSet::new();
+            .iter_streams(prev_stream..)
-
+            .filter(|s| !s.is_null_stream())
-        for stream in self.indices.streams.range(prev_stream..).skip(1).cloned() {
+            .skip(1)
-            if !stream.is_null_stream() {
+            .next();
-                next_stream = Some(stream);
-                break;
-            } else {
-                null_streams.insert(stream);
-            }
-        }
-
-        self.indices.streams = self.indices.streams.sub(&null_streams);
 
         if let Some(next_stream) = next_stream {
             let var = self.deref_register(2).as_var().unwrap();
@@ -3772,7 +3754,7 @@ impl Machine {
     pub(crate) fn flush_output(&mut self) -> CallResult {
         let mut stream = self.machine_st.get_stream_or_alias(
             self.machine_st.registers[1],
-            &self.indices.stream_aliases,
+            &self.indices,
             atom!("flush_output"),
             1,
         )?;
@@ -3859,7 +3841,7 @@ impl Machine {
     pub(crate) fn close(&mut self) -> CallResult {
         let mut stream = self.machine_st.get_stream_or_alias(
             self.machine_st.registers[1],
-            &self.indices.stream_aliases,
+            &self.indices,
             atom!("close"),
             2,
         )?;
@@ -3868,47 +3850,22 @@ impl Machine {
             stream.flush().unwrap(); // 8.11.6.1b)
         }
 
-        self.indices.streams.remove(&stream);
+        if stream == self.user_input || stream == self.user_output || stream.is_stderr() {
-
+            // stdin, stdout and stderr shouldn't be removed from the store, so return now
-        if stream == self.user_input {
+            return Ok(());
-            self.user_input = self
-                .indices
-                .stream_aliases
-                .get(&atom!("user_input"))
-                .cloned()
-                .unwrap();
-
-            self.indices.streams.insert(self.user_input);
-        } else if stream == self.user_output {
-            self.user_output = self
-                .indices
-                .stream_aliases
-                .get(&atom!("user_output"))
-                .cloned()
-                .unwrap();
-
-            self.indices.streams.insert(self.user_output);
         }
 
-        if !stream.is_stdin() && !stream.is_stdout() && !stream.is_stderr() {
+        self.indices.remove_stream(stream);
-            if let Some(alias) = stream.options().get_alias() {
-                self.indices.stream_aliases.swap_remove(&alias);
-            }
 
-            let close_result = stream.close();
+        stream.close().map_err(|_| {
+            let stub = functor_stub(atom!("close"), 1);
+            let addr = stream_as_cell!(stream);
+            let err = self
+                .machine_st
+                .existence_error(ExistenceError::Stream(addr));
 
-            if close_result.is_err() {
+            self.machine_st.error_form(err, stub)
-                let stub = functor_stub(atom!("close"), 1);
+        })
-                let addr = stream_as_cell!(stream);
-                let err = self
-                    .machine_st
-                    .existence_error(ExistenceError::Stream(addr));
-
-                return Err(self.machine_st.error_form(err, stub));
-            }
-        }
-
-        Ok(())
     }
 
     #[inline(always)]
@@ -4466,11 +4423,10 @@ impl Machine {
                         &mut self.machine_st.arena,
                     );
                     *stream.options_mut() = StreamOptions::default();
-                    if let Some(alias) = stream.options().get_alias() {
-                        self.indices.stream_aliases.insert(alias, stream);
-                    }
 
-                    self.indices.streams.insert(stream);
+                    self.indices
+                        .add_stream(stream, atom!("http_open"), 3)
+                        .map_err(|stub_gen| stub_gen(&mut self.machine_st))?;
 
                     let stream = stream_as_cell!(stream);
 
@@ -4688,7 +4644,10 @@ impl Machine {
                 );
                 *stream.options_mut() = StreamOptions::default();
                 stream.options_mut().set_stream_type(StreamType::Binary);
-                self.indices.streams.insert(stream);
+
+                self.indices.add_stream(stream, atom!("http_accept"), 7)
+                    .map_err(|stub_gen| stub_gen(&mut self.machine_st))?;
+
                 let stream = stream_as_cell!(stream);
 
                                 let handle: TypedArenaPtr<HttpResponse> = arena_alloc!(request.response, &mut self.machine_st.arena);
@@ -4802,7 +4761,11 @@ impl Machine {
                 );
                 *stream.options_mut() = StreamOptions::default();
                 stream.options_mut().set_stream_type(StreamType::Binary);
-                self.indices.streams.insert(stream);
+
+
+                self.indices.add_stream(stream, atom!("http_answer"), 4)
+                    .map_err(|stub_gen| stub_gen(&mut self.machine_st))?;
+
                 let stream = stream_as_cell!(stream);
                 self.machine_st.bind(stream_addr.as_var().unwrap(), stream);
                 }
@@ -5117,11 +5080,10 @@ impl Machine {
                     .stream_from_file_spec(file_spec, &mut self.indices, &options)?;
 
             *stream.options_mut() = options;
-            self.indices.streams.insert(stream);
 
-            if let Some(alias) = stream.options().get_alias() {
+            self.indices
-                self.indices.stream_aliases.insert(alias, stream);
+                .add_stream(stream, atom!("open"), 4)
-            }
+                .map_err(|stub_gen| stub_gen(&mut self.machine_st))?;
 
             let stream_var = self.deref_register(3);
             self.machine_st
@@ -5200,9 +5162,9 @@ impl Machine {
 
     #[inline(always)]
     pub(crate) fn set_stream_options(&mut self) -> CallResult {
-        let mut stream = self.machine_st.get_stream_or_alias(
+        let stream = self.machine_st.get_stream_or_alias(
             self.machine_st.registers[1],
-            &self.indices.stream_aliases,
+            &self.indices,
             atom!("open"),
             4,
         )?;
@@ -5212,10 +5174,12 @@ impl Machine {
         let reposition = self.machine_st.registers[4];
         let stream_type = self.machine_st.registers[5];
 
-        let options =
+        let new_options =
             self.machine_st
                 .get_stream_options(alias, eof_action, reposition, stream_type);
-        *stream.options_mut() = options;
+        self.indices.update_stream_options(stream, |options| {
+            *options = new_options;
+        });
 
         Ok(())
     }
@@ -5958,12 +5922,9 @@ impl Machine {
     pub(crate) fn set_input(&mut self) -> CallResult {
         let addr = self.deref_register(1);
 
-        let stream = self.machine_st.get_stream_or_alias(
+        let stream =
-            addr,
+            self.machine_st
-            &self.indices.stream_aliases,
+                .get_stream_or_alias(addr, &self.indices, atom!("set_input"), 1)?;
-            atom!("set_input"),
-            1,
-        )?;
 
         if !stream.is_input_stream() {
             let stub = functor_stub(atom!("set_input"), 1);
@@ -5979,18 +5940,16 @@ impl Machine {
         }
 
         self.user_input = stream;
+        self.indices.set_stream(atom!("user_input"), stream);
         Ok(())
     }
 
     #[inline(always)]
     pub(crate) fn set_output(&mut self) -> CallResult {
         let addr = self.deref_register(1);
-        let stream = self.machine_st.get_stream_or_alias(
+        let stream =
-            addr,
+            self.machine_st
-            &self.indices.stream_aliases,
+                .get_stream_or_alias(addr, &self.indices, atom!("set_output"), 1)?;
-            atom!("set_output"),
-            1,
-        )?;
 
         if !stream.is_output_stream() {
             let stub = functor_stub(atom!("set_output"), 1);
@@ -6006,6 +5965,7 @@ impl Machine {
         }
 
         self.user_output = stream;
+        self.indices.set_stream(atom!("user_output"), stream);
         Ok(())
     }
 
@@ -6296,7 +6256,7 @@ impl Machine {
 
         let stream = self.machine_st.get_stream_or_alias(
             self.machine_st.registers[1],
-            &self.indices.stream_aliases,
+            &self.indices,
             atom!("read_term"),
             3,
         )?;
@@ -6537,7 +6497,7 @@ impl Machine {
         }
 
         if let Some(alias) = options.get_alias() {
-            if self.indices.stream_aliases.contains_key(&alias) {
+            if self.indices.has_stream(alias) {
                 return Err(self.machine_st.occupied_alias_permission_error(
                     alias,
                     atom!("socket_client_open"),
@@ -6553,11 +6513,9 @@ impl Machine {
 
                 *stream.options_mut() = options;
 
-                if let Some(alias) = stream.options().get_alias() {
+                self.indices
-                    self.indices.stream_aliases.insert(alias, stream);
+                    .add_stream(stream, atom!("socket_client_open"), 7)
-                }
+                    .map_err(|stub_gen| stub_gen(&mut self.machine_st))?;
-
-                self.indices.streams.insert(stream);
 
                 stream_as_cell!(stream)
             }
@@ -6565,7 +6523,7 @@ impl Machine {
                 return Err(self.machine_st.open_permission_error(
                     addr,
                     atom!("socket_client_open"),
-                    3,
+                    7,
                 ));
             }
             Err(ErrorKind::NotFound) => {
@@ -6682,7 +6640,7 @@ impl Machine {
         }
 
         if let Some(alias) = options.get_alias() {
-            if self.indices.stream_aliases.contains_key(&alias) {
+            if self.indices.has_stream(alias) {
                 return Err(self.machine_st.occupied_alias_permission_error(
                     alias,
                     atom!("socket_server_accept"),
@@ -6709,11 +6667,10 @@ impl Machine {
 
                                  *tcp_stream.options_mut() = options;
 
-                                 if let Some(alias) = &tcp_stream.options().get_alias() {
+                                 self.indices.add_stream(tcp_stream, atom!("socket_server_accept"), 4)
-                                     self.indices.stream_aliases.insert(*alias, tcp_stream);
+                                    .map_err(|stub_gen| {
-                                 }
+                                        stub_gen(&mut self.machine_st)
-
+                                    })?;
-                                 self.indices.streams.insert(tcp_stream);
 
                                  let tcp_stream = stream_as_cell!(tcp_stream);
                                  let client = atom_as_cell!(client);
@@ -6749,7 +6706,7 @@ impl Machine {
         {
             let stream0 = self.machine_st.get_stream_or_alias(
                 self.machine_st.registers[2],
-                &self.indices.stream_aliases,
+                &self.indices,
                 atom!("tls_client_negotiate"),
                 3,
             )?;
@@ -6768,7 +6725,10 @@ impl Machine {
 
             let addr = atom!("TLS");
             let stream = Stream::from_tls_stream(addr, stream, &mut self.machine_st.arena);
-            self.indices.streams.insert(stream);
+
+            self.indices
+                .add_stream(stream, atom!("tls_client_negotiate"), 3)
+                .map_err(|stub_gen| stub_gen(&mut self.machine_st))?;
 
             self.machine_st.heap.push(stream_as_cell!(stream));
             let stream_addr = self.deref_register(3);
@@ -6803,7 +6763,7 @@ impl Machine {
 
             let stream0 = self.machine_st.get_stream_or_alias(
                 self.machine_st.registers[3],
-                &self.indices.stream_aliases,
+                &self.indices,
                 atom!("tls_server_negotiate"),
                 3,
             )?;
@@ -6822,7 +6782,10 @@ impl Machine {
             };
 
             let stream = Stream::from_tls_stream(atom!("TLS"), stream, &mut self.machine_st.arena);
-            self.indices.streams.insert(stream);
+
+            self.indices
+                .add_stream(stream, atom!("tls_server_negotiate"), 3)
+                .map_err(|stub_gen| stub_gen(&mut self.machine_st))?;
 
             let stream_addr = self.deref_register(4);
             self.machine_st
@@ -6864,7 +6827,7 @@ impl Machine {
     pub(crate) fn set_stream_position(&mut self) -> CallResult {
         let mut stream = self.machine_st.get_stream_or_alias(
             self.machine_st.registers[1],
-            &self.indices.stream_aliases,
+            &self.indices,
             atom!("set_stream_position"),
             2,
         )?;
@@ -6907,7 +6870,7 @@ impl Machine {
     pub(crate) fn stream_property(&mut self) -> CallResult {
         let mut stream = self.machine_st.get_stream_or_alias(
             self.machine_st.registers[1],
-            &self.indices.stream_aliases,
+            &self.indices,
             atom!("stream_property"),
             2,
         )?;
@@ -7258,7 +7221,7 @@ impl Machine {
     pub(crate) fn write_term(&mut self) -> CallResult {
         let mut stream = self.machine_st.get_stream_or_alias(
             self.machine_st.registers[1],
-            &self.indices.stream_aliases,
+            &self.indices,
             atom!("write_term"),
             3,
         )?;
@@ -8134,7 +8097,7 @@ impl Machine {
     pub(crate) fn devour_whitespace(&mut self) -> CallResult {
         let mut stream = self.machine_st.get_stream_or_alias(
             self.machine_st.registers[1],
-            &self.indices.stream_aliases,
+            &self.indices,
             atom!("$devour_whitespace"),
             1,
         )?;

src/tests/stream-aliasing.pl

@@ -0,0 +1,19 @@
+% NOTE: the tests in this file will need to be changed once
+% `open(stream(S1), read, _, NewOptions)` creates a new stream handle `S2` instead of updating
+% the options in `S1`. In that case, that specific query will need to be updated
+% to instead change the options of `S1`.
+
+alias_dropped_stream :-
+    open("README.md", read, S, [alias(readme)]),
+    open(stream(S), read, _, [alias(not_readme)]),
+    close(S),
+    stream_property(readme, file_name(_)). % Should throw an existence_error
+
+realias_user_output :-
+    current_output(S),
+    open(stream(S), read, _, [alias(not_user_output)]),
+    stream_property(S, alias(user_output)). % Should succeed
+
+set_output_alias :-
+    set_output(user_error),
+    write(user_output, hello). % Should write into stderr, not stdout

tests/scryer/cli/src_tests/alias_dropped_stream.stdin

@@ -0,0 +1,2 @@
+alias_dropped_stream.
+halt.

tests/scryer/cli/src_tests/alias_dropped_stream.stdout

@@ -0,0 +1 @@
+   error(existence_error(stream,readme),stream_property/0).

tests/scryer/cli/src_tests/alias_dropped_stream.toml

@@ -0,0 +1,2 @@
+# Part of issue 2806
+args = ["-f", "--no-add-history", "src/tests/stream-aliasing.pl"]

tests/scryer/cli/src_tests/realias_user_output.stdin

@@ -0,0 +1,2 @@
+realias_user_output.
+halt.

tests/scryer/cli/src_tests/realias_user_output.stdout

@@ -0,0 +1 @@
+   true.

tests/scryer/cli/src_tests/realias_user_output.toml

@@ -0,0 +1,2 @@
+# Part of issue 2806
+args = ["-f", "--no-add-history", "src/tests/stream-aliasing.pl"]

tests/scryer/cli/src_tests/set_output_alias.stderr

@@ -0,0 +1 @@
+hello   true.

tests/scryer/cli/src_tests/set_output_alias.stdin

@@ -0,0 +1,2 @@
+set_output_alias.
+halt.

tests/scryer/cli/src_tests/set_output_alias.toml

@@ -0,0 +1,2 @@
+# Part of issue 2806
+args = ["-f", "--no-add-history", "src/tests/stream-aliasing.pl"]